Mercurial > code > home > repos > homeauto
annotate service/mqtt_to_rdf/mqtt_message.py @ 1732:3f4b447d65f5
port to starlette/asyncio
| author | drewp@bigasterisk.com |
|---|---|
| date | Mon, 10 Jul 2023 17:37:58 -0700 |
| parents | 2085ed9cfcc4 |
| children |
| rev | line source |
|---|---|
| 1630 | 1 import json |
| 2 import uuid | |
| 3 | |
| 4 from rdflib import RDF, URIRef, BNode, Graph, Literal, Namespace | |
| 5 from rdflib.collection import Collection | |
| 6 | |
| 7 ROOM = Namespace('http://projects.bigasterisk.com/room/') | |
| 8 JSON = Namespace('http://bigasterisk.com/anyJson/') | |
| 9 | |
| 10 | |
|
1706
2085ed9cfcc4
reworking UI to reflect the new inferencing code
drewp@bigasterisk.com
parents:
1630
diff
changeset
|
11 def graphFromMessage(topicUri: URIRef, topic: bytes, body: bytes): |
| 1630 | 12 graph = Graph() |
| 13 message = URIRef(f'{uuid.uuid1().urn}') | |
| 14 | |
|
1706
2085ed9cfcc4
reworking UI to reflect the new inferencing code
drewp@bigasterisk.com
parents:
1630
diff
changeset
|
15 graph.add((topicUri, ROOM['message'], message)) |
|
2085ed9cfcc4
reworking UI to reflect the new inferencing code
drewp@bigasterisk.com
parents:
1630
diff
changeset
|
16 |
| 1630 | 17 graph.add((message, RDF.type, ROOM['MqttMessage'])) |
| 18 | |
| 19 topicSegments = BNode() | |
| 20 graph.add((message, ROOM['topic'], topicSegments)) | |
| 21 Collection(graph, topicSegments, map(Literal, topic.decode('ascii').split('/'))) | |
| 22 | |
| 23 bodyStr = body.decode('utf8') | |
| 24 graph.add((message, ROOM['body'], Literal(bodyStr))) | |
| 25 try: | |
| 26 graph.add((message, ROOM['bodyFloat'], Literal(float(bodyStr)))) | |
| 27 except ValueError: | |
| 28 pass | |
| 29 _maybeAddJson(graph, message, bodyStr) | |
| 30 return graph | |
| 31 | |
| 32 | |
| 33 def _maybeAddJson(graph, message, bodyStr): | |
| 34 if not bodyStr.startswith('{'): | |
| 35 return | |
| 36 try: | |
| 37 doc = json.loads(bodyStr) | |
| 38 except ValueError: | |
| 39 return | |
| 40 print(f'got {doc=}') | |
| 41 jsonRoot = BNode() | |
| 42 graph.add((message, ROOM['bodyJson'], jsonRoot)) | |
| 43 for k, v in doc.items(): | |
| 44 graph.add((jsonRoot, JSON[k], Literal(v))) |