first commit

Logger.py

@@ -0,0 +1,18 @@
+
+#from Worker import Worker
+
+class Logger(object):
+
+    log = open("/home/alko/Applications/i3statusbar/status1.log", 'w')  
+#    worker = Worker()
+
+    @staticmethod
+    def logMessage(message):
+        Logger._writeMessage(message)
+        #Logger.worker.addJob(Logger._writeMessage, message)
+
+    @staticmethod
+    def _writeMessage(message):
+        Logger.log.write(message + "\n")
+        Logger.log.flush()
+

PulseClient.py

@@ -0,0 +1,163 @@
+
+from threading import Lock 
+from pulsectl import Pulse, PulseLoopStop
+
+from Logger import Logger
+from Worker import Worker
+
+class Singleton(type):
+    _instances = {}
+    def __call__(cls, *args, **kwargs):
+        if cls not in cls._instances:
+            cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
+        return cls._instances[cls]
+
+class PulseClient(object):
+    __metaclass__ = Singleton
+
+    def __init__(self):
+        self.pulseCallLock = Lock()
+        self.pulseListenLock = Lock()
+
+        self.pulse = Pulse('PulseClient')
+        self.pulse.module_load('module-switch-on-connect')
+
+        self.pulse.event_callback_set(self.pulseEventCallback)
+        self.pulse.event_mask_set('sink', 'source', 'card')
+        #self.pulse.event_mask_set('all')
+
+        self.sinkEventCallback = None
+        self.sourceEventCallback = None 
+
+        self.worker = Worker()
+        self.worker.setIdleCallback(self.onWorkerIdle)
+        self.onWorkerIdle()
+        self.worker.start()
+
+    def stop(self):
+        pass
+
+    def setSinkEventCallback(self, callback):
+        self.sinkEventCallback = callback
+
+    def setSourceEventCallback(self, callback):
+        self.sourceEventCallback = callback
+
+    def onWorkerIdle(self, **args):
+        #Logger.logMessage("workerIdleCallback!!!")
+        self.worker.addJob(self.pulseListen)
+
+    def pulseListen(self, **args):
+        self.pulseCallLock.acquire()
+        self.pulseListenLock.acquire()
+        self.pulseCallLock.release()
+        self.pulse.event_listen()
+        self.pulseListenLock.release()
+
+    def pulseEventCallback(self, event):
+        #Logger.logMessage("pulseEventCallback!!!: {0}".format(event))
+        if event.facility == "sink":
+            self.worker.addJob(self.sinkEventCallback, facility='sink')
+        elif event.facility == "source":
+            self.worker.addJob(self.sourceEventCallback, facility='source')
+        else:
+            self.worker.addJob(self.muteAll)
+        raise PulseLoopStop
+
+    def _doPulseCall(self, func, **args):
+        self.pulseCallLock.acquire()
+        self.pulse.event_listen_stop()
+        self.pulseListenLock.acquire()
+        self.pulseListenLock.release()
+        result = func(**args)
+        self.pulseCallLock.release()
+        return result
+
+    def getDefaultSink(self):
+        for sink in self.pulse.sink_list():
+            if sink.name == self.pulse.server_info().default_sink_name:
+                return sink
+
+    def getDefaultSource(self):
+        for source in self.pulse.source_list():
+            if source.name == self.pulse.server_info().default_source_name:
+                return source
+
+    def _doMuteAllSinks(self, **args):
+        for sink in self.pulse.sink_list():
+            self.pulse.mute(sink, mute = True)
+
+    def muteAllSinks(self):
+        self._doPulseCall(self._doMuteAllSinks)
+
+    def _doMuteAllSources(self, **args):
+        for source in self.pulse.source_list():
+            self.pulse.mute(source, mute = True)
+
+    def muteAllSources(self):
+        self._doPulseCall(self._doMuteAllSources)
+
+    def muteAll(self):
+        self.muteAllSinks()
+        self.muteAllSources()
+
+    #####
+
+    def _doGetDefaultSinkMute(self, **args):
+        #Logger.logMessage("sink mute={0}".format(self.getDefaultSink().mute))
+        return self.getDefaultSink().mute == 1
+
+    def getDefaultSinkMute(self):
+        return self._doPulseCall(self._doGetDefaultSinkMute)
+
+    def _doSetDefaultSinkMute(self, **args):
+        #Logger.logMessage("_doSetDefaultSinkMute")
+        self.pulse.mute(self.getDefaultSink(), mute=args["mute"])
+
+    def setDefaultSinkMute(self, mute):
+        self._doPulseCall(self._doSetDefaultSinkMute, mute=mute)
+
+    #####
+
+    def _doGetDefaultSourceMute(self, **args):
+        #Logger.logMessage("source  mute={0}".format(self.getDefaultSource().mute))
+        return self.getDefaultSource().mute == 1
+
+    def getDefaultSourceMute(self):
+        return self._doPulseCall(self._doGetDefaultSourceMute)
+
+    def _doSetDefaultSourceMute(self, **args):
+        #Logger.logMessage("_doSetDefaultSourceMute")
+        self.pulse.mute(self.getDefaultSource(), mute=args["mute"])
+
+    def setDefaultSourceMute(self, mute):
+        self._doPulseCall(self._doSetDefaultSourceMute, mute=mute)
+
+    #####
+
+    def _doGetDefaultSinkVolume(self, **args):
+        return self.pulse.volume_get_all_chans(self.getDefaultSink())
+
+    def getDefaultSinkVolume(self):
+        return self._doPulseCall(self._doGetDefaultSinkVolume)
+
+    def _doSetDefaultSinkVolume(self, **args):
+        self.pulse.volume_set_all_chans(self.getDefaultSink(), args["value"])
+
+    def setDefaultSinkVolume(self, value):
+        self._doPulseCall(self._doSetDefaultSinkVolume, value=value)
+
+    #####
+
+    def _doGetDefaultSourceVolume(self, **args):
+        return self.pulse.volume_get_all_chans(self.getDefaultSource())
+
+    def getDefaultSourceVolume(self):
+        return self._doPulseCall(self._doGetDefaultSourceVolume)
+
+    def _doSetDefaultSourceVolume(self, **args):
+        self.pulse.volume_set_all_chans(self.getDefaultSource(), args["value"])
+
+    def setDefaultSourceVolume(self, value):
+        self._doPulseCall(self._doSetDefaultSourceVolume, value=value)
+

README.md

@@ -0,0 +1 @@
+# i3statusbar

StatusLine.py

@@ -0,0 +1,74 @@
+from threading import Timer
+
+from Logger import Logger
+
+class StatusLine(object):
+    def __init__(self, refreshCallback):
+        self.lastClickedControl = None
+        self.controls = {}
+        self.orderedNames = []
+        self.refreshCallback = refreshCallback
+
+        self.timer = None
+
+    def start(self):
+        pass
+
+    def stop(self):
+        #TODO: stop all controls
+        if self.timer != None and self.timer.is_alive():
+            self.timer.cancel()
+
+    def pause(self):
+        pass
+
+    def resume(slef):
+        pass
+
+    def addControl(self, control):
+        name = control.name
+        control.setRefreshCallback(self.onControlChange)
+        self.orderedNames.append(name)
+        self.controls[name] = control
+        self.doUpdate()
+
+    def doUpdate(self):
+        #Logger.logMessage("update line")
+        self.refreshCallback(list(self.blocks))
+
+    def onControlChange(self):
+        # do not update more then 10 times in a second
+        if self.timer == None or not self.timer.is_alive():
+            self.timer = Timer(0.1, self.doUpdate)
+            self.timer.start()
+
+    @property
+    def blocks(self):
+        for name in self.orderedNames:
+            for block in self.controls[name].blocks:
+                yield block.getAttrs()
+
+    def processEvent(self, event):
+        #Logger.logMessage("click event")
+
+        if not event.has_key("name") or not self.controls.has_key(event["name"]):
+            return
+
+
+        clickedControl = self.controls[event["name"]]
+
+        if self.lastClickedControl == None:
+            self.lastClickedControl = clickedControl
+            #self.lastClickedControl.isActive = True
+        else:
+            if self.lastClickedControl.name != clickedControl.name:
+                #Logger.logMessage("change active control")
+                self.lastClickedControl.isActive = False
+                self.lastClickedControl = clickedControl
+                #self.lastClickedControl.isActive = True
+
+        self.lastClickedControl.clicked(event)
+
+    def refresh(self):
+        for name, control in self.controls.items():
+            control.update()

StatusLineAC.py

@@ -0,0 +1,44 @@
+# -*- coding: utf-8 -*-
+
+from subprocess import check_output, Popen, PIPE
+import shlex
+import colorsys
+import pyudev
+
+from Logger import Logger
+from StatusLineControl import StatusLineControl
+from StatusLineBlock import StatusLineBlock
+
+class StatusLineAC(StatusLineControl):
+
+    def __init__(self):
+        StatusLineControl.__init__(self, "AC")
+
+        self.acLabel = StatusLineBlock("🔌")
+
+        self.udevCtx = pyudev.Context()
+        self.udevMon = pyudev.Monitor.from_netlink(self.udevCtx)
+        self.udevMon.filter_by(subsystem='power_supply')
+        self.udevObs = pyudev.MonitorObserver(self.udevMon, callback=self.onUdevEvent)
+
+        for device in self.udevCtx.list_devices(subsystem='power_supply'):
+            self.onUdevEvent(device)
+
+        self.udevObs.start()
+
+    @property
+    def blocks(self):
+        if self.isACOnline:
+            yield self.acLabel
+
+    def onUdevEvent(self, device):
+        #Logger.logMessage(device.sys_name)
+        if device.sys_name == self.name:
+            self.isACOnline = device.attributes.asbool('online')
+            self.update()
+
+    def doOnUpdate(self):
+        pass
+
+    def doOnUpdateDone(self):
+        pass

StatusLineBattery.py

@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+
+from subprocess import check_output, Popen, PIPE
+import shlex
+import colorsys
+import pyudev
+
+from Logger import Logger
+from StatusLineControl import StatusLineControl
+from StatusLineBlock import StatusLineBlock
+
+# 🔋
+
+class StatusLineBattery(StatusLineControl):
+
+    def __init__(self):
+        StatusLineControl.__init__(self, "BAT")
+
+        self.bat = {}
+        self.bat["BAT0"] = StatusLineBlock("BAT0 --.--%")
+        self.bat["BAT1"] = StatusLineBlock("🔋 --.--%")
+
+        self.udevCtx = pyudev.Context()
+        self.udevMon = pyudev.Monitor.from_netlink(self.udevCtx)
+        self.udevMon.filter_by(subsystem='power_supply')
+        self.udevObs = pyudev.MonitorObserver(self.udevMon, callback=self.onUdevEvent)
+
+        for device in self.udevCtx.list_devices(subsystem='power_supply'): 
+            self.onUdevEvent(device)
+
+        self.udevObs.start()
+
+    @property
+    def blocks(self):
+        yield self.bat["BAT0"]
+        yield self.bat["BAT1"]
+
+    def getBatteryCharge(self, device):
+        now = device.attributes.asint('energy_now')
+        full = device.attributes.asint('energy_full')
+        return float(now)/full*100
+
+    def onUdevEvent(self, device):
+        if device.sys_name == "BAT0" or device.sys_name == "BAT1":
+            charge = self.getBatteryCharge(device)
+            color = colorsys.hsv_to_rgb(1.0/360*120*(charge)/100, 1, 1)
+            color = [int(c *255) for c in color]
+            self.bat[device.sys_name].color = "#{0[0]:0>2x}{0[1]:0>2x}{0[2]:0>2x}".format(color)
+            self.bat[device.sys_name].full_text = "{0} {1:.2f}%".format(device.sys_name, charge)
+
+    def doOnUpdate(self):
+        pass
+
+    def doOnUpdateDone(self):
+        pass