-
Notifications
You must be signed in to change notification settings - Fork 9
/
server.py
129 lines (93 loc) · 3.43 KB
/
server.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
#
# uchroma - Copyright (C) 2017 Steve Kondik
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, version 3.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
#
import argparse
import asyncio
import atexit
import logging
import signal
from concurrent import futures
import gbulb
from uchroma.log import Log, LOG_PROTOCOL_TRACE, LOG_TRACE
from uchroma.util import ensure_future
from .dbus import DeviceManagerAPI
from .device_manager import UChromaDeviceManager
from .power import PowerMonitor
class UChromaServer:
def __init__(self):
gbulb.install()
parser = argparse.ArgumentParser(description='UChroma daemon')
parser.add_argument('-v', "--version", action='version', version='self.version')
parser.add_argument('-d', "--debug", action='append_const', const=True,
help="Increase logging verbosity")
parser.add_argument('-C', "--colorlog", action='store_true',
help="Use colored log output")
args = parser.parse_args()
self._loop = asyncio.get_event_loop()
level = logging.INFO
asyncio_debug = False
colorlog = False
if args.colorlog is not None:
colorlog = args.colorlog
Log.enable_color(colorlog)
self._logger = Log.get('uchroma.server')
if args.debug is not None:
if len(args.debug) > 2:
level = LOG_PROTOCOL_TRACE
asyncio_debug = True
elif len(args.debug) == 2:
level = LOG_TRACE
asyncio_debug = True
elif len(args.debug) == 1:
level = logging.DEBUG
logging.getLogger().setLevel(level)
self._loop.set_debug(asyncio_debug)
def _shutdown_callback(self):
self._logger.info("Shutting down")
self._loop.stop()
def run(self):
try:
self._run()
except KeyboardInterrupt:
pass
def _run(self):
dm = UChromaDeviceManager()
atexit.register(UChromaServer.exit, self._loop)
dbus = DeviceManagerAPI(dm, self._logger)
power = PowerMonitor()
for sig in (signal.SIGINT, signal.SIGTERM):
self._loop.add_signal_handler(sig, self._shutdown_callback)
try:
dbus.run()
power.start()
ensure_future(dm.monitor_start(), loop=self._loop)
self._loop.run_forever()
except KeyboardInterrupt:
pass
finally:
for sig in (signal.SIGTERM, signal.SIGINT):
self._loop.remove_signal_handler(sig)
power.stop()
self._loop.run_until_complete(asyncio.wait( \
[dm.close_devices(), dm.monitor_stop()],
return_when=futures.ALL_COMPLETED))
@staticmethod
def exit(loop):
try:
loop.run_until_complete(asyncio.wait( \
list(asyncio.Task.all_tasks()),
return_when=futures.ALL_COMPLETED))
loop.close()
except KeyboardInterrupt:
pass
def run_server():
UChromaServer().run()