Purge action manager

action-manager/actions.py

@@ -1,207 +0,0 @@
-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

action-manager/command.sh

@@ -1,5 +0,0 @@
-#!/bin/bash
-cat /dev/stdin | $(cat /dev/stdin > "$1")&
-quit_proc="$!"
-sleep 1
-kill "$quit_proc"

action-manager/daemon.py

@@ -1,73 +0,0 @@
-#!/usr/bin/env python3
-import argparse
-import time
-import actions
-import pickle
-
-class CreateFileType(argparse.FileType):
-    def __init__(self, mode='r', bufsize=-1, encoding=None, errors=None):
-        super().__init__(mode, bufsize, encoding, errors)
-        self.__mode = mode
-        self.__bufsize = bufsize
-        self.__encoding = encoding
-        self.__errors = errors
-
-    def __call__(self, string):
-        try:
-            return super().__call__(string)
-        except argparse.ArgumentTypeError as e:
-            return open(string, 'wb' if 'b' in self.__mode else 'w', self.__bufsize, self.__encoding, self.__errors)
-
-
-
-parser = argparse.ArgumentParser(description='Pulseaudio volume manager for xmobar')
-parser.add_argument('output_pipe', type=argparse.FileType('w',bufsize=1))
-parser.add_argument('command_pipe', type=argparse.FileType('r',bufsize=1))
-parser.add_argument('--state-file', type=CreateFileType('r+b'))
-
-modules = [
-    actions.QuitControl(),
-    actions.ChildReaperControl(),
-    actions.RedshiftControl(),
-    actions.VolumeControl(),
-]
-
-[m.configure(parser.add_argument_group(m.__class__.__name__)) for m in modules]
-args = parser.parse_args()
-[m.bind_arguments(args) for m in modules]
-
-if args.state_file is not None and args.state_file.readable():
-    try:
-        state = pickle.load(args.state_file)
-        print(state)
-        [m.load_state(state[m.__class__.__name__]) for m in modules if m.enabled and m.__class__.__name__ in state]
-    except Exception as e:
-        print(e)
-
-
-def output_pipe(fd):
-    bars=' | '.join([str(m) for m in modules if m.visible])
-    fd.writelines(bars+"\n")
-
-def command_pipe(fd):
-    command = str.rstrip(fd.readline())
-    if len(command) == 0:
-        return False
-    return any([m.respond_to(command) for m in modules if m.enabled])
-
-
-try: 
-    while True:
-        if command_pipe(args.command_pipe) or any([m.periodic() for m in modules if m.enabled]):
-            output_pipe(args.output_pipe)
-        else:
-            time.sleep(1)
-finally:
-    if args.state_file is not None:
-        args.state_file.seek(0)
-        args.state_file.truncate()
-        state = {m.__class__.__name__: m.dump_state() for m in modules if m.enabled}
-        pickle.dump(state, args.state_file)
-        args.state_file.close()
-
-    [m.cleanup() for m in modules if m.enabled]

action-manager/functions.py

@@ -1,21 +0,0 @@
-import itertools
-import math
-
-def action(command, text, **kwargs):
-    return '<action=`{}`{}>{}</action>'.format(command,' '+(' '.join(['{}={}'.format(k, v) for k,v in kwargs.items()])) if len(kwargs) else '', text)
-
-def create_bars(volume):
-    num_bars=float(volume)/9000.0
-    return ('/'*math.floor(num_bars))+partial_bar(num_bars-math.floor(num_bars))+(' '*(10-math.ceil(num_bars)))
-
-def partial_bar(bar_size):
-    if bar_size == 0.0:
-        return ''
-    elif bar_size < 0.3:
-        return ' '
-    elif bar_size < 0.6:
-        return '.'
-    elif bar_size < 0.9:
-        return '-'
-    return '/'
-

action-manager/start.sh

@@ -1,36 +0,0 @@
-#!/bin/bash
-ARGS="$@"
-while [[ -n "$@" ]]; do
-    if [[ "${1:0:1}" != "-" ]]; then
-        if [[ -z "$volumepipe" ]]; then
-            volumepipe="$1"
-        elif [[ -z "$commandpipe" ]]; then
-            commandpipe="$1"
-        fi
-    fi
-    shift;
-done
-
-
-
-# volumepipe
-if ! [[ -p "$volumepipe" ]]; then
-    rm -rf "$volumepipe"
-    mkfifo "$volumepipe"
-fi
-
-# volumecontrol
-if [[ -p "$commandpipe" ]]; then
-    $(echo "q" > $commandpipe)&
-    quit_proc=$!
-    sleep 1
-    kill $quit_proc
-    rm "$commandpipe"
-else
-    rm -f "$commandpipe"
-fi
-
-mkfifo "$commandpipe"
-$(sleep 1; echo "refresh" > $commandpipe)&
-eval set -- "$ARGS"
-exec "$(dirname "${BASH_SOURCE[0]}")/daemon.py" "$@"