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
|
"""
This plugin will set your status to **away** if you detach your screen.
The default behaviour is to check for both tmux and screen (in that order).
Configuration options
---------------------
.. glossary::
use_screen
**Default:** ``true``
Try to find an attached screen.
use_tmux
**Default:** ``true``
Try to find and attached tmux.
use_csi
**Default:** ``false``
Use `client state indication`_ to limit bandwidth (thus CPU) usage when detached. WARNING: using CSI together with chatrooms will result in inaccurate logs due to presence filtering or other inaccuracies.
.. _client state indication: https://xmpp.org/extensions/xep-0352.html
"""
from poezio.plugin import BasePlugin
import os
import stat
import pyinotify
import asyncio
DEFAULT_CONFIG = {
'screen_detach': {
'use_tmux': True,
'use_screen': True,
'use_csi': False
}
}
# overload if this is not how your stuff
# is configured
try:
LOGIN = os.getlogin() or ''
LOGIN_TMUX = os.getuid()
except Exception:
LOGIN = os.getenv('USER') or ''
LOGIN_TMUX = os.getuid()
SCREEN_DIR = '/var/run/screens/S-%s' % LOGIN
TMUX_DIR = '/tmp/tmux-%s' % LOGIN_TMUX
def find_screen(path):
if not os.path.isdir(path):
return
for f in os.listdir(path):
path = os.path.join(path, f)
if screen_attached(path):
return path
def screen_attached(socket):
return (os.stat(socket).st_mode & stat.S_IXUSR) != 0
class Plugin(BasePlugin, pyinotify.Notifier):
default_config = DEFAULT_CONFIG
def init(self):
sock_path = None
if self.config.get('use_tmux'):
sock_path = find_screen(TMUX_DIR)
if sock_path is None and self.config.get('use_screen'):
sock_path = find_screen(SCREEN_DIR)
# Only actually do something if we found an attached screen (assuming only one)
if sock_path:
self.attached = True
wm = pyinotify.WatchManager()
wm.add_watch(sock_path,
pyinotify.EventsCodes.ALL_FLAGS['IN_ATTRIB'])
pyinotify.Notifier.__init__(
self, wm, default_proc_fun=HandleScreen(plugin=self))
asyncio.get_event_loop().add_reader(self._fd, self.process)
else:
self.api.information(
'screen_detach plugin: No tmux or screen found', 'Warning')
self.attached = False
def process(self):
self.read_events()
self.process_events()
def cleanup(self):
asyncio.get_event_loop().remove_reader(self._fd)
def update_screen_state(self, socket):
attached = screen_attached(socket)
if attached != self.attached:
self.attached = attached
status = 'available' if self.attached else 'away'
self.core.command.status(status)
if self.config.get('use_csi'):
if self.attached:
self.core.xmpp.plugin['xep_0352'].send_active()
else:
self.core.xmpp.plugin['xep_0352'].send_inactive()
class HandleScreen(pyinotify.ProcessEvent):
def my_init(self, **kwargs):
self.plugin = kwargs['plugin']
def process_IN_ATTRIB(self, event):
self.plugin.update_screen_state(event.path)
|
