xmonad/action-manager/actions.py

import functions
import subprocess
import abc
import sys
import os.path

class AbstractControl(metaclass=abc.ABCMeta):
    def __init__(self):
        self.args = None

    @property
    def visible(self):
        return self.enabled

    @property
    def enabled(self):
        return True

    def configure(self, argument_parser):
        pass

    def bind_arguments(self, args):
        self.args = args

    def periodic(self):
        pass

    def cleanup(self):
        pass

    def respond_to(self, command):
        return False

    def load_state(self, state):
        pass

    def dump_state(self):
        return None

    def create_pipe_command(self, command):
        return 'echo {} | {}/command.sh {}'.format(command, os.path.abspath(sys.path[0]), os.path.abspath(self.args.command_pipe.name))
        


class VolumeControl(AbstractControl):

    @property
    def muted(self):
        return self._muted

    @muted.setter
    def muted(self, muted):
        if self._muted != muted:
            try:
                self._muted = muted
                self._pactl('set-sink-mute', str(int(muted)))
            except subprocess.CalledProcessError as e:
                pass

    @property
    def volume(self):
        return self._volume

    @volume.setter
    def volume(self, volume):
        if self.muted:
            self.muted = False
        if self._volume != volume:
            try:
                self._volume = volume
                self._pactl('set-sink-volume', str(volume))
            except subprocess.CalledProcessError as e:
                pass

    def _pactl(self, command, arg):
        for i in range(6):
            subprocess.check_call(["pactl", command, str(i), arg])


    def __init__(self):
        super().__init__()
        self._muted = False
        self._volume = 0

    def respond_to(self, command):
        if command[0] == '=':
            self.volume = int(command[1:])*9000
        elif command == 'm1':
            self.muted = True
        elif command == 'm0':
            self.muted = False
        elif command == 'mt':
            self.muted = not self.muted
        elif command == '+':
            self.volume+=3000
        elif command == '-':
            self.volume-=3000
        elif command == 'r':
            self.volume=30000
        else:
            return False
        return True

    def __str__(self):
        return self.action_bars(functions.create_bars(self.volume) if not self.muted else '  (mute)  ')

    def action_bars(self, bars):
        return ''.join([functions.action(self.create_pipe_command('=%d'%(i+1)), c, button=1) for i,c in zip(range(len(bars)), bars)])

    def load_state(self, state):
        self.volume = state['volume']
        self.muted = state['muted']

    def dump_state(self):
        return dict(volume=self.volume, muted=self.muted)

class QuitControl(AbstractControl):
    @property
    def visible(self):
        return False

    def respond_to(self, command):
        if command == 'q':
            sys.exit(0)
        elif command == 'refresh':
            return True

class RedshiftControl(AbstractControl):
    def __init__(self):
        super().__init__()
        self._redshift_proc = None

    def configure(self, argument_parser):
        argument_parser.add_argument('--redshift-enabled', help='Use the redshift module', action='store_true')
        argument_parser.add_argument('--redshift-location', help='LAT:LON Your current location', type=str)
        argument_parser.add_argument('--redshift-temperature', help='DAY:NIGHT Color temperature to set at daytime/night', type=str)

    def bind_arguments(self, args):
        super().bind_arguments(args)
        if self.enabled and not self.redshift_error_message:
            self.redshift_enabled = True

    @property
    def enabled(self):
        return self.args.redshift_enabled

    @property
    def redshift_enabled(self):
        return bool(self._redshift_proc)

    @property
    def redshift_error_message(self):
        if self.enabled and not (self.args.redshift_location or self.args.redshift_temperature):
            return "Missing parameter(s) --redshift-location and/or --redshift-temperature"
        if self._redshift_proc is not None and self._redshift_proc.returncode is not None and self._redshift_proc.returncode != 0:
            return self._redshift_proc.communicate()[1].replace("\n", ' ')
        return None


    @redshift_enabled.setter
    def redshift_enabled(self, value):
        if value == self.redshift_enabled:
            return
        if value:
            self._redshift_proc = subprocess.Popen(['redshift', '-l', self.args.redshift_location, '-t', self.args.redshift_temperature], )
        else:
            self._redshift_proc.terminate()

    def periodic(self):
        if self._redshift_proc:
            self._redshift_proc.poll()
            if self._redshift_proc.returncode is not None:
                self._redshift_proc = None
                return True
        
    def respond_to(self, command):
        if command == 'redshift':
            self.redshift_enabled = not self.redshift_enabled
            return True
        return False

    def cleanup(self):
        self.redshift_enabled = False
        if self._redshift_proc:
            self._redshift_proc.wait()

    def __str__(self):
        if not self.redshift_error_message:
            return functions.action(self.create_pipe_command('redshift'), 'R' if self.redshift_enabled else 'r')
        return 'E: '+self.redshift_error_message

    def load_state(self, state):
        self.redshift_enabled = state['redshift_enabled']

    def dump_state(self):
        return {'redshift_enabled': self.redshift_enabled}

class ChildReaperControl(AbstractControl):
    @property
    def visible(self):
        return False

    def periodic(self):
        try:
            os.wait3(os.WNOHANG)
        except:
            pass