path: root/tp-key-handler

#!/usr/bin/python3
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
import dbus
import dbus.service
import subprocess
import time
from dbus.mainloop.glib import DBusGMainLoop
import notify2
import sys
from time import gmtime, strftime

class KeyhandlerDBUSService(dbus.service.Object):
    def __init__(self):
        bus_name = dbus.service.BusName('org.elektranox.keyhandler', bus=dbus.SessionBus(), allow_replacement=True , replace_existing=True)
        dbus.service.Object.__init__(self, bus_name, '/org/elektranox/keyhandler')
        self.bnotice = notify2.Notification("Backlight", "", "display-brightness-symbolic")
        self.vnotice = notify2.Notification("Volume", "", "")
        self.wnotice = notify2.Notification("WLAN", "", "")
        self.mnotice = notify2.Notification("Mouse", "", "")
        self.timeout = 2000

    def _check_sway(self):
        # sway is not running in X
        popen = subprocess.Popen(['/usr/bin/swaymsg', '-q'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        popen.wait()
        return not popen.returncode

    def _lid_status_open(self):
        lidstatus = "state:      unknown"
        with open('/proc/acpi/button/lid/LID/state', 'r') as lidstatefile:
            lidstatus = lidstatefile.read()
        lidstatus = lidstatus.strip().replace("state:      ", "")
        return lidstatus == "open"

    def _dock_status_docked(self):
        popen = subprocess.Popen(['/usr/bin/lsusb', '-d', '17ef:1010'])
        popen.wait()
        return not popen.returncode

    def _bluetooth_set_status(self, state):
        bus = dbus.SystemBus()
        dev = bus.get_object("org.bluez", u'/org/bluez/hci0')
        iface = dbus.Interface(dev, 'org.freedesktop.DBus.Properties')
        iface.Set("org.bluez.Adapter1", "Powered", state)

    def _bluetooth_get_status(self, state):
        bus = dbus.SystemBus()
        dev = bus.get_object("org.bluez", u'/org/bluez/hci0')
        iface = dbus.Interface(dev, 'org.freedesktop.DBus.Properties')
        return iface.Get("org.bluez.Adapter1", "Powered")

    def _getBacklight(self):
        with open("/sys/class/backlight/intel_backlight/brightness", "r") as f:
            return int(f.readline().strip())

    def _getMaxBacklight(self):
        with open("/sys/class/backlight/intel_backlight/max_brightness", "r") as f:
            return int(f.readline().strip())

    def _setBacklight(self, value):
        with open("/sys/class/backlight/intel_backlight/brightness", "w") as f:
            f.write(str(value))

    def _showBacklightState(self):
        state = self._getBacklight() * 100 / self._getMaxBacklight()
        self.bnotice.set_hint_int32("value", int(state))
        self.bnotice.set_hint_string("synchronous", "brightness")
        self.bnotice.timeout = self.timeout
        try:
            self.bnotice.show()
        except BaseException as e:
            print(e)
            Gtk.main_quit()

    def _showVolumeState(self):
        muted = False
        popen = subprocess.Popen(['/usr/bin/pulsemixer', '--get-mute'], stdout = subprocess.PIPE)
        popen.wait()
        (stdout,stderr) = popen.communicate(None)
        if "1" in stdout.decode("utf-8"):
            muted = True
        popen = subprocess.Popen(['/usr/bin/pulsemixer', '--get-volume'], stdout = subprocess.PIPE)
        popen.wait()
        (stdout,stderr) = popen.communicate(None)
        channels = stdout.decode("utf-8").split("\n")[0].split(" ")
        state = int((int(channels[0]) + int(channels[1])) / 2)
        if muted:
            self.vnotice.icon = "audio-volume-muted"
        elif state < 20:
            self.vnotice.icon = "audio-volume-low"
        elif state < 60:
            self.vnotice.icon = "audio-volume-medium"
        else:
            self.vnotice.icon = "audio-volume-high"
        self.vnotice.set_hint_int32("value", state)
        self.vnotice.set_hint_string("synchronous", "volume")
        self.vnotice.timeout = self.timeout
        try:
            self.vnotice.show()
        except BaseException as e:
            print(e)
            Gtk.main_quit()

    def _rfkill_get(self, rfid):
        popen = subprocess.Popen(['/usr/sbin/rfkill', 'list', rfid, '-o', 'soft', '-rn'], stdout = subprocess.PIPE)
        popen.wait()
        (stdout,stderr) = popen.communicate()
        state = stdout.decode("utf-8").split("\n")[0].strip()
        return state

    def _showMicState(self):
        popen = subprocess.Popen(['/usr/bin/amixer', '-c', 'PCH', 'get', 'Capture'], stdout = subprocess.PIPE)
        popen.wait()
        (stdout,stderr) = popen.communicate(None)
        lines = stdout.decode("utf-8").split("\n")
        state = int(lines[-2].split()[4][1:-2])
        muted = lines[-2].split()[6][1:-1] == "off"
        if muted:
            self.vnotice.icon = "microphone-sensitivity-muted"
        elif state < 20:
            self.vnotice.icon = "microphone-sensitivity-low"
        elif state < 60:
            self.vnotice.icon = "microphone-sensitivity-medium"
        else:
            self.vnotice.icon = "microphone-sensitivity-high"
        self.vnotice.set_hint_int32("value", state)
        self.vnotice.set_hint_string("synchronous", "volume")
        self.vnotice.timeout = self.timeout
        try:
            self.vnotice.show()
        except BaseException as e:
            print(e)
            Gtk.main_quit()

    @dbus.service.method('org.elektranox.keyhandler')
    def volumeMute(self):
        popen = subprocess.Popen(['/usr/bin/pulsemixer', '--toggle-mute'])
        popen.wait()
        self._showVolumeState()

    @dbus.service.method('org.elektranox.keyhandler')
    def volumeDec(self):
        popen = subprocess.Popen(['/usr/bin/pulsemixer', '--change-volume', '-2'], stdout=subprocess.DEVNULL)
        popen.wait()
        self._showVolumeState()

    @dbus.service.method('org.elektranox.keyhandler')
    def volumeInc(self):
        popen = subprocess.Popen(['/usr/bin/pulsemixer', '--change-volume', '+2'], stdout=subprocess.DEVNULL)
        popen.wait()
        self._showVolumeState()

    @dbus.service.method('org.elektranox.keyhandler')
    def micMute(self):
        popen = subprocess.Popen(['/usr/bin/amixer', 'set', 'Capture', 'toggle'], stdout=subprocess.DEVNULL)
        popen.wait()
        self._showMicState()

    def _backlightChg(self, decrease):
        brightness = self._getBacklight()
        maximum = self._getMaxBacklight()
        current = self._getBacklight() / maximum * 100
        step = 1

        # probably should implement something using a logarithmic scale
        if current <= 2:
            step = int(0.5 * maximum/100)
        elif current <= 10:
            step = int(1 * maximum/100)
        elif current <= 50:
            step = int(5 * maximum/100)
        else:
            step = int(10 * maximum/100)

        if decrease:
            brightness -= step
        else:
            brightness += step

        if brightness < 0:
            brightness = 0
        if brightness > maximum:
            brightness = maximum
        self._setBacklight(brightness)
        self._showBacklightState()

    @dbus.service.method('org.elektranox.keyhandler')
    def backlightDec(self):
        self._backlightChg(True)

    @dbus.service.method('org.elektranox.keyhandler')
    def backlightInc(self):
        self._backlightChg(False)

    @dbus.service.method('org.elektranox.keyhandler')
    def XF86Display(self):
        lidopen = self._lid_status_open()
        docked = self._dock_status_docked()

        if self._check_sway():
            if not docked:
                popen = subprocess.Popen(['/usr/bin/swaymsg','output', 'eDP-1', 'enable'])
                popen.wait()
                popen = subprocess.Popen(['/usr/bin/notify-send','Not docked'])
                popen.wait()
            elif lidopen:
                popen = subprocess.Popen(['/usr/bin/swaymsg','output', 'eDP-1', 'enable'])
                popen.wait()
                popen = subprocess.Popen(['/usr/bin/swaymsg','output', 'DP-3', 'enable'])
                popen.wait()
                popen = subprocess.Popen(['/usr/bin/swaymsg','output', 'eDP-1', 'pos', '0', '0'])
                popen.wait()
                popen = subprocess.Popen(['/usr/bin/swaymsg','output', 'DP-3', 'pos', '1920', '0'])
                popen.wait()
                popen = subprocess.Popen(['/usr/bin/notify-send','Docked with open lid'])
                popen.wait()
            else:
                popen = subprocess.Popen(['/usr/bin/swaymsg','output', 'eDP-1', 'disable'])
                popen.wait()
                popen = subprocess.Popen(['/usr/bin/notify-send','Docked with closed lid'])
                popen.wait()
        else:
            if not docked:
                popen = subprocess.Popen(['/usr/bin/xrandr','--output', 'eDP1', '--auto'])
                popen.wait()
                popen = subprocess.Popen(['/usr/bin/xrandr','--output', 'DP2-1', '--off'])
                popen.wait()
                popen = subprocess.Popen(['/usr/bin/notify-send','Not docked'])
                popen.wait()
            elif lidopen:
                popen = subprocess.Popen(['/usr/bin/xrandr','--output', 'eDP1', '--auto'])
                popen.wait()
                popen = subprocess.Popen(['/usr/bin/xrandr','--output', 'DP2-1', '--right-of', 'eDP1', '--auto'])
                popen.wait()
                popen = subprocess.Popen(['/usr/bin/notify-send','Docked with open lid'])
                popen.wait()
            else:
                popen = subprocess.Popen(['/usr/bin/xrandr','--output', 'DP2-1', '--right-of', 'eDP1', '--auto'])
                popen.wait()
                popen = subprocess.Popen(['/usr/bin/xrandr','--output', 'eDP1', '--off'])
                popen.wait()
                popen = subprocess.Popen(['/usr/bin/notify-send','Docked with closed lid'])
                popen.wait()

    @dbus.service.method('org.elektranox.keyhandler')
    def WLAN(self):
        # button itself is handled by systemd
        state = self._rfkill_get("wlan")
        if state == "blocked":
                subprocess.call(['/usr/sbin/rfkill','block', 'wlan'])
                self.wnotice.icon = "network-wireless-offline-symbolic"
                self.wnotice.summary = "WLAN disabled"
        else:
                subprocess.call(['/usr/sbin/rfkill','unblock', 'wlan'])
                self.wnotice.icon = "network-wireless-symbolic"
                self.wnotice.summary = "WLAN enabled"
        self.wnotice.timeout = self.timeout
        self.wnotice.show()

    @dbus.service.method('org.elektranox.keyhandler')
    def XF86Tools(self):
        popen = subprocess.Popen(['/usr/local/bin/toggle-mouse-wheel-emulation'], stdout = subprocess.PIPE)
        popen.wait()
        (stdout,stderr) = popen.communicate(None)
        enabled = (stdout.decode("utf-8").find("enabled") != -1)
        self.mnotice.icon = "/usr/share/icons/Adwaita/48x48/devices/input-mouse-symbolic.symbolic.png"
        if enabled:
                self.mnotice.summary = "Mouse Wheel Emulation enabled"
        else:
                self.mnotice.summary = "Mouse Wheel Emulation disabled"
        self.mnotice.timeout = self.timeout
        try:
            self.mnotice.show()
        except BaseException as e:
            print(e)
            Gtk.main_quit()

    @dbus.service.method('org.elektranox.keyhandler')
    def XF86Search(self):
        popen = subprocess.Popen(['/usr/bin/notify-send','XF86Search'])
        popen.wait()
        # TODO: add some fancy binding
        pass

    @dbus.service.method('org.elektranox.keyhandler')
    def XF86LaunchA(self):
        timestamp = strftime("%Y-%m-%d_%H:%M:%S", gmtime())
        filename = "screenshot-" + timestamp + ".png"
        popen = subprocess.Popen(['/usr/bin/grim', filename])
        popen.wait()
        popen = subprocess.Popen(['/usr/bin/notify-send','Screenshot: ' + filename])
        popen.wait()
        pass

    @dbus.service.method('org.elektranox.keyhandler')
    def XF86Explorer(self):
        state = self._rfkill_get("bluetooth")
        if state == "unblocked":
            subprocess.call(['/usr/sbin/rfkill','block', 'bluetooth'])
            self.wnotice.icon = "/usr/share/icons/Adwaita/48x48/status/bluetooth-disabled-symbolic.symbolic.png"
            self.wnotice.summary = "Bluetooth blocked"
        else:
            subprocess.call(['/usr/sbin/rfkill','unblock', 'bluetooth'])
            self.wnotice.icon = "/usr/share/icons/Adwaita/48x48/status/bluetooth-active-symbolic.symbolic.png"
            self.wnotice.summary = "Bluetooth unblocked"
        self.wnotice.timeout = self.timeout
        try:
            self.wnotice.show()
        except BaseException as e:
            print(e)
            Gtk.main_quit()
        if not state == "unblocked":
            time.sleep(1)
            self._bluetooth_set_status(True)

def main():
    notify2.init("TP Keyhandler", mainloop='glib')
    service = KeyhandlerDBUSService()
    Gtk.main()
    service.remove_from_connection()

if __name__ == '__main__':
    while True:
        main()