Mercurial > code > home > repos > environment

comparison environment.py @ 0:3c1bc3bc5a6c

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
pull out of homeauto/ project, and add skaffold/pipenv
author drewp@bigasterisk.com
date Thu, 31 Mar 2022 22:03:43 -0700
parents
children 0f532eb91364
comparison
equal deleted inserted replaced
-1:000000000000 0:3c1bc3bc5a6c
1 #!/usr/bin/python
2 """
3 return some rdf about the environment, e.g. the current time,
4 daytime/night, overall modes like 'maintenance mode', etc
5
6 """
7 import datetime
8
9 import cyclone.web
10 from dateutil.relativedelta import FR, relativedelta
11 from dateutil.tz import tzlocal
12 from docopt import docopt
13 from rdflib import Literal, Namespace
14 from twisted.internet import defer, reactor, task
15
16 from standardservice.logsetup import log, verboseLogging
17 from patchablegraph import (CycloneGraphEventsHandler, CycloneGraphHandler, PatchableGraph)
18
19 from patch_cyclone import patch_sse
20 from rdfdoc import Doc
21 from twilight import isWithinTwilight
22
23 patch_sse()
24
25 ROOM = Namespace("http://projects.bigasterisk.com/room/")
26 DEV = Namespace("http://projects.bigasterisk.com/device/")
27
28 # STATS = scales.collection(
29 # '/root',
30 # scales.PmfStat('update'),
31 # )
32
33
34 # see pending fix in patchablegraph.py
35 class CycloneGraphEventsHandlerWithCors(CycloneGraphEventsHandler):
36
37 def __init__(self, application, request, masterGraph, allowOrigins=None):
38 super().__init__(application, request, masterGraph)
39 self.allowOrigins = allowOrigins or []
40
41 def flush(self):
42 origin = self.request.headers.get('origin', None)
43 if origin and origin in self.allowOrigins:
44 self.set_header(b"Access-Control-Allow-Origin", origin.encode('utf8'))
45 self.set_header(b"Access-Control-Allow-Credentials", b"true")
46 return CycloneGraphEventsHandler.flush(self)
47
48
49 # @STATS.update.time()
50 def update(masterGraph):
51 stmt = lambda s, p, o: masterGraph.patchObject(ROOM.environment, s, p, o)
52
53 now = datetime.datetime.now(tzlocal())
54
55 stmt(DEV.environment, ROOM.localHour, Literal(now.hour))
56 stmt(DEV.environment, ROOM.localTimeToMinute, Literal(now.strftime("%H:%M")))
57
58 stmt(DEV.environment, ROOM.localTimeToSecond, Literal(now.strftime("%H:%M:%S")))
59
60 stmt(DEV.environment, ROOM.localDayOfWeek, Literal(now.strftime("%A")))
61 stmt(DEV.environment, ROOM.localMonthDay, Literal(now.strftime("%B %e")))
62 stmt(DEV.environment, ROOM.localDate, Literal(now.strftime("%Y-%m-%d")))
63
64 for offset in range(-12, 7):
65 d = now.date() + datetime.timedelta(days=offset)
66 if d == d + relativedelta(day=31, weekday=FR(-1)):
67 stmt(DEV.calendar, ROOM.daysToLastFridayOfMonth, Literal(offset))
68
69 stmt(DEV.calendar, ROOM.twilight, ROOM['withinTwilight'] if isWithinTwilight(now) else ROOM['daytime'])
70
71
72 def main():
73 arg = docopt("""
74 Usage: environment.py [options]
75
76 -v Verbose
77 """)
78 verboseLogging(arg['-v'])
79
80 masterGraph = PatchableGraph()
81
82 class Application(cyclone.web.Application):
83
84 def __init__(self):
85 handlers = [
86 (r"/()", cyclone.web.StaticFileHandler, {
87 "path": ".",
88 "default_filename": "index.html"
89 }),
90 (r'/graph/environment', CycloneGraphHandler, {
91 'masterGraph': masterGraph
92 }),
93 (r'/graph/environment/events', CycloneGraphEventsHandlerWithCors, {
94 'masterGraph': masterGraph,
95 'allowOrigins': ['http://localhost:8001'],
96 }),
97 (r'/doc', Doc), # to be shared
98 # (r'/stats/(.*)', StatsHandler, {
99 # 'serverName': 'environment'
100 # }),
101 ]
102 cyclone.web.Application.__init__(self, handlers, masterGraph=masterGraph)
103
104 task.LoopingCall(update, masterGraph).start(1)
105 reactor.listenTCP(8080, Application())
106 reactor.run()
107
108
109 if __name__ == '__main__':
110 main()