view environment.py @ 5:b5bfd0dd69d6

upadte bg loop, patchablegraph, py version
author drewp@bigasterisk.com
date Sun, 28 Aug 2022 21:37:20 -0700
parents e7f33fa31883
children 145779f2d79d
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 datetime
import logging

import background_loop
from dateutil.relativedelta import FR, relativedelta
from dateutil.tz import tzlocal
from patchablegraph import PatchableGraph
from patchablegraph.handler import GraphEvents, StaticGraph
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

# 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)
logging.getLogger('patchablegraph').setLevel(logging.WARNING)


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()

    loop = background_loop.loop_forever(lambda first_run=False: update(masterGraph), 1)

    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()