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
|
#!/usr/bin/env python3
import sys
import dbus
import re
import datetime
# import pprint
class AutoSuspender(object):
def __init__(self):
self.__bus = dbus.SystemBus()
self.__timer_re = re.compile('^syncoid-run-.*\.timer$')
# self.__pp = pprint.PrettyPrinter(indent=2)
def _get_interface(self, dest, object, interface):
try:
obj = self.__bus.get_object(dest, object)
return dbus.Interface(obj, interface)
except dbus.exceptions.DBusException as error:
print(error)
sys.exit(1)
def _get_logind_manager_interface(self):
return self._get_interface("org.freedesktop.login1", "/org/freedesktop/login1", "org.freedesktop.login1.Manager")
def _get_systemd_manager_interface(self):
return self._get_interface("org.freedesktop.systemd1", "/org/freedesktop/systemd1", "org.freedesktop.systemd1.Manager")
def _get_systemd_timer_properties(self, path):
return self._get_interface("org.freedesktop.systemd1", path, "org.freedesktop.DBus.Properties").GetAll("org.freedesktop.systemd1.Timer")
def _get_users_logged_in(self):
manager = self._get_logind_manager_interface()
users = []
for user in manager.ListUsers():
uid = int(user[0])
name = str(user[1])
users.append((name, uid))
return users
def _get_syncoid_timers(self):
manager = self._get_systemd_manager_interface()
timers = []
for unit in manager.ListUnits():
name = str(unit[0])
path = str(unit[6])
if not self.__timer_re.match(name):
continue
timers.append((name, path))
return timers
def _get_timer_next_elapse(self, path):
props = self._get_systemd_timer_properties(path)
try:
return datetime.datetime.fromtimestamp(int(props['NextElapseUSecRealtime']) / 1000000)
except ValueError:
return datetime.datetime.now()
def check(self):
result = True
users = self._get_users_logged_in()
if(len(users) > 0):
print("%d users logged in: %s" % (len(users), ", ".join([user[0] for user in users])))
result = False
else:
print("no users logged in.")
timers = self._get_syncoid_timers()
for timer in timers:
next = self._get_timer_next_elapse(timer[1])
until = next - datetime.datetime.now()
if(until < datetime.timedelta(minutes=10)):
print("timer '%s' elapses in less then 10 Minutes -> %s (%s)" % (timer[0], next, until))
result = False
else:
print("timer '%s' elapses in %s (%s)" % (timer[0], until, next))
return result
def suspend(self):
manager = self._get_logind_manager_interface()
if not manager.CanSuspend():
print("suspending is not possible!")
return
print("suspending system")
manager.Suspend(False)
if __name__ == "__main__":
s = AutoSuspender()
if s.check():
s.suspend()
else:
print("not suspending system")
|
