Mercurial > code > home > repos > mqtt_metrics

annotate mqtt_metrics.py @ 2:579df3a4e62d

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
rewrite converters as register'able functions
author drewp@bigasterisk.com
date Fri, 09 Aug 2024 17:37:00 -0700
parents 3d7f2dc9beec
children 41e36f98f3b8
Ignore whitespace changes - Everywhere: Within whitespace: At end of lines:
rev   line source
1
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
1 import asyncio
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
2 import json
0
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
3 import logging
2
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
4 import re
1
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
5 import time
2
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
6 from typing import cast
1
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
7 from weakref import WeakSet
0
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
8
1
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
9 import aiomqtt
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
10 import uvicorn # v 2.0.0
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
11 from sse_starlette.sse import EventSourceResponse
0
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
12 from starlette.applications import Starlette
2
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
13 from starlette.routing import Route
0
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
14 from starlette.requests import Request
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
15 from starlette_exporter import PrometheusMiddleware, handle_metrics
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
16
2
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
17 from convert import converters
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
18
0
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
19 logging.basicConfig(level=logging.INFO)
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
20 log = logging.getLogger()
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
21
1
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
22 debugListeners = WeakSet()
0
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
23
1
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
24
2
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
25 def broadcastToDebugListeners(event: dict):
1
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
26 j = json.dumps(event)
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
27 for lis in debugListeners:
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
28 lis.put_nowait(j)
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
29
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
30
2
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
31 async def debugEvents(request: Request) -> EventSourceResponse:
1
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
32 q = asyncio.Queue()
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
33 debugListeners.add(q)
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
34
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
35 async def gen():
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
36 try:
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
37 while True:
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
38 yield await q.get()
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
39 except asyncio.CancelledError:
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
40 debugListeners.discard(q)
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
41
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
42 resp = EventSourceResponse(gen())
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
43 return resp
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
44
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
45
2
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
46 def simplifyMqttMessage(mqttMessage: aiomqtt.Message) -> dict:
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
47 message = {
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
48 't': round(time.time(), 3),
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
49 'topic': mqttMessage.topic.value,
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
50 'payload': mqttMessage.payload,
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
51 }
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
52 if isinstance(message['payload'], bytes):
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
53 message['payload'] = message['payload'].decode('utf-8')
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
54
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
55 return message
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
56
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
57
1
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
58 async def mqttTask():
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
59 client = aiomqtt.Client('mqtt2.bigasterisk.com', identifier="mqtt-exporter")
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
60 async with client:
2
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
61 await client.subscribe('#')
1
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
62 async for mqttMessage in client.messages:
2
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
63 onMqttMessage(mqttMessage)
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
64
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
65
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
66 def onMqttMessage(mqttMessage):
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
67 message = simplifyMqttMessage(mqttMessage)
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
68
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
69 metricEvent = tryConverters(message)
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
70 if metricEvent:
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
71 broadcastToDebugListeners({'message': message, 'metricEvent': metricEvent})
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
72
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
73
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
74 def tryConverters(message) -> dict | None:
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
75 metricEvent = None
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
76 matchedPats = []
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
77 for converter in converters:
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
78 if not (m := re.search(converter.topic_pattern, message['topic'])):
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
79 continue
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
80 if not matchedPats:
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
81 try:
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
82 metricEvent = converter(message, cast(tuple[str], m.groups()))
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
83 except Exception:
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
84 log.error("Error converting mqtt message: topic pattern = %r", converter.topic_pattern, exc_info=True)
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
85 matchedPats.append(converter.topic_pattern)
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
86
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
87 if len(matchedPats) > 1:
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
88 log.warning("Multiple patterns matched: %s", ", ".join(matchedPats))
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
89
579df3a4e62d rewrite converters as register'able functions
drewp@bigasterisk.com
parents: 1
diff changeset
90 return metricEvent
0
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
91
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
92
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
93 def main():
1
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
94 loop = asyncio.new_event_loop()
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
95 app = Starlette(on_startup=[lambda: loop.create_task(mqttTask())], debug=True, routes=[
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
96 Route('/api/debugEvents', debugEvents),
0
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
97 ])
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
98 app.add_middleware(PrometheusMiddleware, app_name='mqtt_metrics')
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
99 app.add_route("/metrics", handle_metrics)
1
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
100 config = uvicorn.Config(app=app, host='0.0.0.0', port=8001, loop='asyncio')
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
101 server = uvicorn.Server(config)
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
102 loop.run_until_complete(server.serve())
0
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
103
0b5b4ede1bf5 start with a mockup of the debugging display
drewp@bigasterisk.com
parents:
diff changeset
104
1
3d7f2dc9beec read 1 mqtt topic; dummy convert; route to debugging view
drewp@bigasterisk.com
parents: 0
diff changeset
105 main()