homeauto: service/mqtt_to_rdf/mqtt_to

comparison service/mqtt_to_rdf/mqtt_to_rdf.py @ 1629:1c36ad1eb8b3

do inference on config. backend for new ui columns. rm some of the old filter pipeline

author	drewp@bigasterisk.com
date	Sat, 11 Sep 2021 23:31:32 -0700
parents	b0608eb6e90c
children	78024b27f9ec

comparison

equal deleted inserted replaced

-:24e8cd8fcdcd
+:1c36ad1eb8b3
 """
 Subscribe to mqtt topics; generate RDF statements.
 """
+import glob
+import json
+import logging
+import os
+from mqtt_message import graphFromMessage
 import os
 import time
-import json
+from dataclasses import dataclass
-import logging
 from pathlib import Path
 from typing import Callable, Sequence, Set, Tuple, Union, cast
-from cyclone.util import ObjectDict
-from rdflib.graph import ConjunctiveGraph
+import cyclone.sse
-from rx.core.typing import Mapper
-from export_to_influxdb import InfluxExporter
 import cyclone.web
-import cyclone.sse
+import export_to_influxdb
 import prometheus_client
 import rx
 import rx.operators
 import rx.scheduler.eventloop
 from docopt import docopt
+from export_to_influxdb import InfluxExporter
 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.typing import Mapper
 from standardservice.logsetup import log, verboseLogging
 from twisted.internet import reactor, task
-from dataclasses import dataclass
+from inference import Inference
 from button_events import button_events
 from patch_cyclone_sse import patchCycloneSse
 ROOM = Namespace('http://projects.bigasterisk.com/room/')
 MESSAGES_SEEN = Counter('mqtt_messages_seen', '')
 collectors = {}
-import export_to_influxdb
-print(f'merge me back {export_to_influxdb}')
 patchCycloneSse()
+def logGraph(debug: Callable, label: str, graph: Graph):
+n3 = cast(bytes, graph.serialize(format="n3"))
+debug(label + ':\n' + n3.decode('utf8'))
 def appendLimit(lst, elem, n=10):
 del lst[:len(lst) - n + 1]
 lst.append(elem)
 def tightN3(node: Union[URIRef, Literal]) -> str:
 return node.n3().replace('http://www.w3.org/2001/XMLSchema#', 'xsd:')
-def serializeWithNs(graph: ConjunctiveGraph) -> bytes:
+def serializeWithNs(graph: Graph, hidePrefixes=False) -> str:
 graph.bind('', ROOM)
-return cast(bytes, graph.serialize(format='n3'))
+n3 = cast(bytes, graph.serialize(format='n3')).decode('utf8')
+if hidePrefixes:
+n3 = ''.join(line for line in n3.splitlines(keepends=True) if not line.strip().startswith('@prefix'))
+return n3
 class MqttStatementSource:
 def __init__(self, uri: URIRef, config: Graph, masterGraph: PatchableGraph, mqtt, internalMqtt, debugPageData,
-influxExport: InfluxExporter):
+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.inference = inference
 self.mqttTopic = self.topicFromConfig(self.config)
 log.debug(f'new mqttTopic {self.mqttTopic}')
 self.debugSub = {
 'topic': self.mqttTopic.decode('ascii'),
-'recentMessages': [],
+'recentMessageGraphs': [],
-'recentParsed': [],
+'recentMetrics': [],
-'recentConversions': [],
-'currentMetrics': [],
 'currentOutputGraph': {
 't': 1,
 'n3': "(n3)"
 },
 }
 self.debugPageData['subscribed'].append(self.debugSub)
 rawBytes: Observable = self.subscribeMqtt(self.mqttTopic)
 rawBytes.subscribe(on_next=self.countIncomingMessage)
-filteredBytes = Filters(uri, config).makeOutputStream(rawBytes)
+rawBytes.subscribe_(self.onMessage)
-parsedObjs = Parser(uri, config).makeOutputStream(filteredBytes)
+def onMessage(self, raw: bytes):
-parsedObjs.subscribe_(lambda v: appendLimit(self.debugSub['recentParsed'], {'t': truncTime(), 'n3': tightN3(v)}))
+g = graphFromMessage(self.mqttTopic, raw)
+logGraph(log.debug, 'message graph', g)
-convertedObjs = Converters(uri, config).makeOutputStream(parsedObjs)
+appendLimit(
-convertedObjs.subscribe_(lambda v: appendLimit(self.debugSub['recentConversions'], {'t': truncTime(), 'n3': tightN3(v)}))
+self.debugSub['recentMessageGraphs'],
+{  #
-outputQuadsSets = Rdfizer(uri, config).makeOutputStream(convertedObjs)
+'t': truncTime(),
-outputQuadsSets.subscribe_(self.updateInflux)
+'n3': serializeWithNs(g, hidePrefixes=True)
+})
-outputQuadsSets.subscribe_(self.updateMasterGraph)
+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 countIncomingMessage(self, msg: bytes):
 self.debugPageData['messagesSeen'] += 1
 MESSAGES_SEEN.inc()
-appendLimit(self.debugSub['recentMessages'], {
-'t': truncTime(),
-'msg': msg.decode('ascii'),
-})
 def updateInflux(self, newGraphs):
 for g in newGraphs:
 self.influxExport.exportToInflux(g)
-def updateMasterGraph(self, newGraphs):
+def updateMasterGraph(self, newGraph):
-newQuads = set.union(*newGraphs)
+log.debug(f'{self.uri} update to {len(newGraph)} statements')
-g = graphFromQuads(newQuads)
-log.debug(f'{self.uri} update to {len(newQuads)} statements')
+cg = ConjunctiveGraph()
+for stmt in newGraph:
-for quad in newQuads:
+cg.add(stmt + (self.uri,))
-meas = quad[0].split('/')[-1]
+meas = stmt[0].split('/')[-1]
 if meas.startswith('airQuality'):
 where_prefix, type_ = meas[len('airQuality'):].split('door')
 where = where_prefix + 'door'
 metric = 'air'
 tags = {'loc': where.lower(), 'type': type_.lower()}
-val = quad[2].toPython()
+val = stmt[2].toPython()
 if metric not in collectors:
 collectors[metric] = Gauge(metric, 'measurement', labelnames=tags.keys())
 collectors[metric].labels(**tags).set(val)
-self.masterGraph.patchSubgraph(self.uri, g)
+self.masterGraph.patchSubgraph(self.uri, cg)
-self.debugSub['currentOutputGraph']['n3'] = serializeWithNs(g).decode('utf8')
+self.debugSub['currentOutputGraph']['n3'] = serializeWithNs(cg, hidePrefixes=True)
 class Metrics(cyclone.web.RequestHandler):
 def get(self):
 --cs=STR  Only process config filenames with this substring
 """)
 verboseLogging(arg['-v'])
 logging.getLogger('mqtt').setLevel(logging.INFO)
 logging.getLogger('mqtt_client').setLevel(logging.INFO)
+logging.getLogger('infer').setLevel(logging.INFO)
+log.info('log start')
 config = Graph()
 for fn in Path('.').glob('conf/*.n3'):
 if not arg['--cs'] or str(arg['--cs']) in str(fn):
 log.debug(f'loading {fn}')
 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'])
+inference = Inference()
+inference.setRules(config)
+expandedConfig = inference.infer(config)
+log.info('expanded config:')
+for stmt in sorted(expandedConfig):
+log.info(f'  {stmt}')
 srcs = []
-for src in sorted(config.subjects(RDF.type, ROOM['MqttStatementSource'])):
+for src in sorted(expandedConfig.subjects(RDF.type, ROOM['MqttStatementSource'])):
 srcs.append(
 MqttStatementSource(src,
 config,
 masterGraph,
 mqtt=mqtt,
 internalMqtt=internalMqtt,
 debugPageData=debugPageData,
-influxExport=influxExport))
+influxExport=influxExport,
+inference=inference))
 log.info(f'set up {len(srcs)} sources')
 port = 10018
 reactor.listenTCP(port,
 cyclone.web.Application([

Mercurial > code > home > repos > homeauto

comparison service/mqtt_to_rdf/mqtt_to_rdf.py @ 1629:1c36ad1eb8b3