Mercurial > code > home > repos > pi_mqtt

comparison pi_mqtt.py @ 3:0cc41259fddd

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
healthcheck
author drewp@bigasterisk.com
date Sun, 10 Dec 2023 17:26:52 -0800
parents 3989f073ed9e
children 759f0ba7d345
comparison
equal deleted inserted replaced
2:b7b6d7f1ce56 3:0cc41259fddd
5 import background_loop 5 import background_loop
6 import pigpio 6 import pigpio
7 from prometheus_client import Gauge 7 from prometheus_client import Gauge
8 from starlette.applications import Starlette 8 from starlette.applications import Starlette
9 from starlette.requests import Request 9 from starlette.requests import Request
10 from starlette.responses import JSONResponse 10 from starlette.responses import PlainTextResponse
11 from starlette.routing import Route 11 from starlette.routing import Route
12 from starlette_exporter import PrometheusMiddleware, handle_metrics 12 from starlette_exporter import PrometheusMiddleware, handle_metrics
13 13
14 SENSOR_MOTION = Gauge('sensor_motion', 'motion detected', ['loc']) 14 SENSOR_MOTION = Gauge('sensor_motion', 'motion detected', ['loc'])
15 OUTPUT_LIGHT = Gauge('output_light', 'light level', ['loc']) 15 OUTPUT_LIGHT = Gauge('output_light', 'light level', ['loc'])
16 logging.basicConfig(level=logging.INFO) 16 logging.basicConfig(level=logging.INFO)
17 log = logging.getLogger() 17 log = logging.getLogger()
18 18
19 lastSuccessfulPoll = time.time()
19 20
20 def hello(request: Request) -> JSONResponse: 21 def health(request: Request) -> PlainTextResponse:
21 return JSONResponse({"demo": "hello"}) 22 if time.time() - lastSuccessfulPoll > 10:
23 return PlainTextResponse('err', status_code=500)
24 return PlainTextResponse('ok', status_code=200)
22 25
23 26
24 @dataclass 27 @dataclass
25 class Poller: 28 class Poller:
26 garage: pigpio.pi 29 garage: pigpio.pi
27 last_motion = 0 30 last_motion = 0
28 31
29 def poll(self, first_run=False): 32 def poll(self, first_run=False):
33 global lastSuccessfulPoll
30 now = time.time() 34 now = time.time()
31 mo = self.garage.read(4) 35 mo = self.garage.read(4)
32 SENSOR_MOTION.labels(loc='ga').set(mo) 36 SENSOR_MOTION.labels(loc='ga').set(mo)
33 if mo: 37 if mo:
34 self.last_motion = now 38 self.last_motion = now
35 39
36 light_on = self.last_motion > now - 20 * 60 40 light_on = self.last_motion > now - 20 * 60
37 OUTPUT_LIGHT.labels(loc='ga').set(light_on) 41 OUTPUT_LIGHT.labels(loc='ga').set(light_on)
38 self.garage.write(15, not light_on) 42 self.garage.write(15, not light_on)
43 lastSuccessfulPoll = time.time()
39 44
40 45
41 def main(): 46 def main():
42 garage = pigpio.pi(host='garage5', port=8888) 47 garage = pigpio.pi(host='garage5', port=8888)
43 48
45 garage.set_mode(4, pigpio.INPUT) 50 garage.set_mode(4, pigpio.INPUT)
46 garage.set_mode(15, pigpio.OUTPUT) 51 garage.set_mode(15, pigpio.OUTPUT)
47 background_loop.loop_forever(poller.poll, 1) 52 background_loop.loop_forever(poller.poll, 1)
48 53
49 app = Starlette(debug=True, routes=[ 54 app = Starlette(debug=True, routes=[
50 Route('/api/hello', hello), 55 Route('/health', health),
51 ]) 56 ])
52 57
53 app.add_middleware(PrometheusMiddleware, app_name='pi_mqtt') 58 app.add_middleware(PrometheusMiddleware, app_name='pi_mqtt')
54 app.add_route("/metrics", handle_metrics) 59 app.add_route("/metrics", handle_metrics)
55 return app 60 return app