summaryrefslogtreecommitdiff
path: root/plugins/screen_detach.py
blob: 0a2514c4ec78301cc9f8f18a75f2fd37e7fc9ed1 (plain)
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()
    LOGIN_TMUX = os.getuid()
except Exception:
    LOGIN = os.getenv('USER')
    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)