Mercurial > code > home > repos > light-bridge

comparison light_bridge.py @ 7:05fb0319eb64

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
cleanup
author drewp@bigasterisk.com
date Sun, 28 Jan 2024 16:03:40 -0800
parents fc8ed0efcd72
children 140633abfa2a
comparison
equal deleted inserted replaced
6:fc8ed0efcd72 7:05fb0319eb64
1 """ 1 """
2 replaces a lot of mqtt_to_rdf and rdf_to_mqtt 2 replaces a lot of mqtt_to_rdf and rdf_to_mqtt
3 """ 3 """
4 import json 4 import json
5 import logging 5 import logging
6 import time
6 from functools import partial 7 from functools import partial
7 import time
8 8
9 import background_loop
10 from patchablegraph import PatchableGraph 9 from patchablegraph import PatchableGraph
11 from patchablegraph.handler import GraphEvents, StaticGraph 10 from patchablegraph.handler import GraphEvents, StaticGraph
12 from rdflib import Namespace
13 from sse_starlette.sse import EventSourceResponse 11 from sse_starlette.sse import EventSourceResponse
14 from starlette.applications import Starlette 12 from starlette.applications import Starlette
15 from starlette.requests import Request 13 from starlette.requests import Request
16 from starlette.responses import JSONResponse 14 from starlette.responses import JSONResponse
17 from starlette.routing import Route 15 from starlette.routing import Route
18 from starlette_exporter import PrometheusMiddleware, handle_metrics 16 from starlette_exporter import PrometheusMiddleware, handle_metrics
19 17
20 from light import Color, Light, Lights 18 from light import Lights
21
22 EX = Namespace('http://example.com/')
23 19
24 logging.basicConfig(level=logging.INFO) 20 logging.basicConfig(level=logging.INFO)
25 log = logging.getLogger() 21 log = logging.getLogger()
26 22
27 23
50 Route('/api/graph/events', GraphEvents(graph)), 46 Route('/api/graph/events', GraphEvents(graph)),
51 ]) 47 ])
52 48
53 app.add_middleware(PrometheusMiddleware, app_name='light-bridge') 49 app.add_middleware(PrometheusMiddleware, app_name='light-bridge')
54 app.add_route("/metrics", handle_metrics) 50 app.add_route("/metrics", handle_metrics)
55 # app.state.loop = background_loop.loop_forever(lambda first_run=False: update(graph), 1)
56 51
57 return app 52 return app
58 53
59 54
60 app = main() 55 app = main()