Mercurial > code > home > repos > light-bridge
annotate light_bridge.py @ 5:7eeda7f4f9cd
spell it to_dict, for compat with DataClassJsonMixin
| author | drewp@bigasterisk.com |
|---|---|
| date | Sun, 28 Jan 2024 16:01:43 -0800 |
| parents | c6fd04e07db0 |
| children | fc8ed0efcd72 |
| rev | line source |
|---|---|
| 0 | 1 """ |
| 2 replaces a lot of mqtt_to_rdf and rdf_to_mqtt | |
| 3 """ | |
| 4 import json | |
| 5 import logging | |
| 2 | 6 from functools import partial |
| 1 | 7 import time |
| 0 | 8 |
| 9 import background_loop | |
| 10 from patchablegraph import PatchableGraph | |
| 11 from patchablegraph.handler import GraphEvents, StaticGraph | |
| 12 from rdflib import Namespace | |
| 2 | 13 from sse_starlette.sse import EventSourceResponse |
| 0 | 14 from starlette.applications import Starlette |
| 15 from starlette.requests import Request | |
| 16 from starlette.responses import JSONResponse | |
| 17 from starlette.routing import Route | |
| 18 from starlette_exporter import PrometheusMiddleware, handle_metrics | |
| 2 | 19 |
| 20 from light import Color, Light, Lights | |
| 0 | 21 |
| 22 EX = Namespace('http://example.com/') | |
| 23 | |
| 24 logging.basicConfig(level=logging.INFO) | |
| 25 log = logging.getLogger() | |
| 26 | |
| 27 | |
| 28 async def output(lights: Lights, request: Request) -> JSONResponse: | |
| 29 light = lights.byName(request.query_params['light']) | |
|
5
7eeda7f4f9cd
spell it to_dict, for compat with DataClassJsonMixin
drewp@bigasterisk.com
parents:
2
diff
changeset
|
30 return JSONResponse(light.to_dict()) |
| 0 | 31 |
| 32 | |
| 33 async def table(lights: Lights, req: Request) -> EventSourceResponse: | |
| 34 | |
| 35 async def g(): | |
| 36 async for ping in lights.changes(): | |
|
5
7eeda7f4f9cd
spell it to_dict, for compat with DataClassJsonMixin
drewp@bigasterisk.com
parents:
2
diff
changeset
|
37 yield json.dumps({'now': time.time()} | lights.to_dict()) |
| 0 | 38 |
| 39 return EventSourceResponse(g()) | |
| 40 | |
| 41 | |
| 42 def main(): | |
| 43 lights = Lights() | |
| 44 lights.add(Light('do-desk', 'topic1', True, Color('#ff0000'), {'r': 255}, Color('#000000'), 100)) | |
| 45 lights.add(Light('do-desk2', 'topic2', True, Color('#ff00ff'), {'r': 255}, Color('#000000'), 200)) | |
| 46 graph = PatchableGraph() | |
| 47 app = Starlette(debug=True, | |
| 48 routes=[ | |
| 49 Route('/api/output', partial(output, lights), methods=['PUT']), | |
| 50 Route('/api/table', partial(table, lights)), | |
| 51 Route('/api/graph', StaticGraph(graph)), | |
| 52 Route('/api/graph/events', GraphEvents(graph)), | |
| 53 ]) | |
| 54 | |
| 55 app.add_middleware(PrometheusMiddleware, app_name='light-bridge') | |
| 56 app.add_route("/metrics", handle_metrics) | |
| 57 # app.state.loop = background_loop.loop_forever(lambda first_run=False: update(graph), 1) | |
| 58 | |
| 59 return app | |
| 60 | |
| 61 | |
| 62 app = main() |