|
0
|
1 import logging
|
|
|
2 import time
|
|
|
3 from dataclasses import dataclass
|
|
|
4
|
|
|
5 import background_loop
|
|
|
6 import pigpio
|
|
4
|
7 import uvicorn
|
|
0
|
8 from prometheus_client import Gauge
|
|
|
9 from starlette.applications import Starlette
|
|
|
10 from starlette.requests import Request
|
|
3
|
11 from starlette.responses import PlainTextResponse
|
|
0
|
12 from starlette.routing import Route
|
|
|
13 from starlette_exporter import PrometheusMiddleware, handle_metrics
|
|
|
14
|
|
|
15 SENSOR_MOTION = Gauge('sensor_motion', 'motion detected', ['loc'])
|
|
|
16 OUTPUT_LIGHT = Gauge('output_light', 'light level', ['loc'])
|
|
|
17 logging.basicConfig(level=logging.INFO)
|
|
|
18 log = logging.getLogger()
|
|
|
19
|
|
3
|
20 lastSuccessfulPoll = time.time()
|
|
0
|
21
|
|
4
|
22
|
|
3
|
23 def health(request: Request) -> PlainTextResponse:
|
|
|
24 if time.time() - lastSuccessfulPoll > 10:
|
|
5
|
25 log.info('failing health check')
|
|
3
|
26 return PlainTextResponse('err', status_code=500)
|
|
|
27 return PlainTextResponse('ok', status_code=200)
|
|
0
|
28
|
|
|
29
|
|
|
30 @dataclass
|
|
|
31 class Poller:
|
|
|
32 garage: pigpio.pi
|
|
|
33 last_motion = 0
|
|
|
34
|
|
|
35 def poll(self, first_run=False):
|
|
3
|
36 global lastSuccessfulPoll
|
|
0
|
37 now = time.time()
|
|
|
38 mo = self.garage.read(4)
|
|
|
39 SENSOR_MOTION.labels(loc='ga').set(mo)
|
|
|
40 if mo:
|
|
|
41 self.last_motion = now
|
|
|
42
|
|
|
43 light_on = self.last_motion > now - 20 * 60
|
|
|
44 OUTPUT_LIGHT.labels(loc='ga').set(light_on)
|
|
|
45 self.garage.write(15, not light_on)
|
|
3
|
46 lastSuccessfulPoll = time.time()
|
|
0
|
47
|
|
|
48
|
|
|
49 def main():
|
|
7
|
50 host = 'ga-iot5'
|
|
5
|
51 log.info(f'connecting to {host=} - if this hangs, garage5 wg maight be missing ip addr')
|
|
|
52 garage = pigpio.pi(host=host, port=8888)
|
|
|
53 log.info('connected')
|
|
0
|
54
|
|
|
55 poller = Poller(garage)
|
|
|
56 garage.set_mode(4, pigpio.INPUT)
|
|
|
57 garage.set_mode(15, pigpio.OUTPUT)
|
|
|
58 background_loop.loop_forever(poller.poll, 1)
|
|
|
59
|
|
|
60 app = Starlette(debug=True, routes=[
|
|
3
|
61 Route('/health', health),
|
|
0
|
62 ])
|
|
|
63
|
|
|
64 app.add_middleware(PrometheusMiddleware, app_name='pi_mqtt')
|
|
|
65 app.add_route("/metrics", handle_metrics)
|
|
|
66 return app
|
|
|
67
|
|
|
68
|
|
5
|
69 uvicorn.run(main, host="0.0.0.0", port=8001, log_level=logging.DEBUG, factory=True)
|
