Mercurial > code > home > repos > environment
view environment.py @ 3:e7f33fa31883
port to starlette/asyncio
author | drewp@bigasterisk.com |
---|---|
date | Sun, 24 Apr 2022 14:46:32 -0700 |
parents | 0f532eb91364 |
children | b5bfd0dd69d6 |
line wrap: on
line source
#!/usr/bin/python """ return some rdf about the environment, e.g. the current time, daytime/night, overall modes like 'maintenance mode', etc """ import asyncio import datetime import logging from dateutil.relativedelta import FR, relativedelta from dateutil.tz import tzlocal from prometheus_client import Gauge, Summary from rdflib import Literal, Namespace from starlette.applications import Starlette from starlette.routing import Route from starlette.staticfiles import StaticFiles from starlette_exporter import PrometheusMiddleware, handle_metrics import background_loop from patchablegraph import PatchableGraph from patchablegraph.handler import StaticGraph, GraphEvents # from rdfdoc import Doc from twilight import isWithinTwilight ROOM = Namespace("http://projects.bigasterisk.com/room/") DEV = Namespace("http://projects.bigasterisk.com/device/") logging.basicConfig(level=logging.INFO) STAT_UPDATE_UP = Gauge('background_loop_up', 'not erroring') STAT_UPDATE_CALLS = Summary('background_loop_calls', 'calls') def update(masterGraph): def stmt(s, p, o): masterGraph.patchObject(ROOM.environment, s, p, o) now = datetime.datetime.now(tzlocal()) stmt(DEV.environment, ROOM.localHour, Literal(now.hour)) stmt(DEV.environment, ROOM.localTimeToMinute, Literal(now.strftime("%H:%M"))) stmt(DEV.environment, ROOM.localTimeToSecond, Literal(now.strftime("%H:%M:%S"))) stmt(DEV.environment, ROOM.localDayOfWeek, Literal(now.strftime("%A"))) stmt(DEV.environment, ROOM.localMonthDay, Literal(now.strftime("%B %e"))) stmt(DEV.environment, ROOM.localDate, Literal(now.strftime("%Y-%m-%d"))) for offset in range(-12, 7): d = now.date() + datetime.timedelta(days=offset) if d == d + relativedelta(day=31, weekday=FR(-1)): stmt(DEV.calendar, ROOM.daysToLastFridayOfMonth, Literal(offset)) stmt(DEV.calendar, ROOM.twilight, ROOM['withinTwilight'] if isWithinTwilight(now) else ROOM['daytime']) def main(): masterGraph = PatchableGraph() asyncio.create_task(background_loop.loop_forever(lambda first: update(masterGraph), 1, STAT_UPDATE_UP, STAT_UPDATE_CALLS)) app = Starlette( debug=True, routes=[ Route('/', StaticFiles(directory='.', html=True)), Route('/graph/environment', StaticGraph(masterGraph)), Route('/graph/environment/events', GraphEvents(masterGraph)), # Route('/doc', Doc), ]) app.add_middleware(PrometheusMiddleware, app_name='environment') app.add_route("/metrics", handle_metrics) return app app = main()