path: root/xpybar/config/xmonad-monitor.gpp

# -*- python -*-

import os
import sys


from plugins.application import Application
from plugins.clock import Clock
from plugins.cpuinfo import CPUInfo
from plugins.hdparm import HDParm
from plugins.image import Image
from plugins.kmsg import KMsg
from plugins.loadavg import AverageLoad
from plugins.lunar import Lunar
from plugins.menu import Menu
from plugins.ropty import ROPTY
from plugins.solar import Solar

sys.path.append(os.path.dirname(config_file) if '/' in config_file else '.')
from common import *

G.globals         = globals()
G.clock           = Clock(sync_to = 0.5)
%%>if test -x /usr/bin/terminator; then
G.TERMINAL        = 'terminator'
%%>elif test -x /usr/bin/st; then
G.TERMINAL        = 'st'
%%>else
G.TERMINAL        = 'xterm'
%%>fi

G.OUTPUT, G.HEIGHT_PER_LINE, G.YPOS, G.TOP = 0, 12, 0, True
G.FONT = '-misc-fixed-medium-r-normal-*-10-*-*-*-c-*-iso10646-1'

from myalsa       import MyALSA, MixerAction
from mybacklight  import MyBacklight
from mybattery    import MyBattery
from mybrilliance import MyBrilliance
from mycg         import MyCGNegative
from myclock      import MyClock
from mycomputer   import MyComputer
from mycpu        import MyCPU
from myio         import MyIO
from myipaddress  import MyIPAddress
from myirc        import MyIRC
from mylid        import MyLid
from myleapsec    import MyLeapsec
from mymemory     import MyMemory
from mymoc        import MyMOC
from mynetwork    import MyNetwork, Ping
from mynews       import MyNews
from myscroll     import MyScroll
from mystat       import MyStat
from mysun        import MySun
from mytimer      import MyTimer
from mytop        import MyTop
from myweather    import MyWeather
from myxmonad     import MyXMonad


%%>hostname="$(hostname | tr '[[:upper:]]' '[[:lower:]]')"


def mqueue_wait():
    import posix_ipc
    qkey = '/.xpybar.' + os.environ['DISPLAY'].split('.')[0]
    q = posix_ipc.MessageQueue(qkey, posix_ipc.O_CREAT, 0o600, 8, 128)
    while True:
        try:
            message = q.receive(None)[0].decode('utf-8', 'replace').split(' ')
            if message[0] in mqueue_map:
                try:
                    mqueue_map[message[0]](message)
                except Exception as err:
                    print('%s: %s' % (sys.argv[0], str(err)), file = sys.stderr, flush = True)
            else:
                print('%s: unrecognised message: %s' % (sys.argv[0], ' '.join(message)), file = sys.stderr, flush = True)
        except:
            time.sleep(1)



myxmonad = MyXMonad(None)
myscroll = MyScroll(None)
myclock  = MyClock (lambda f : Clocked(f, 1), format = '%Y-(%m)%b-%d %T, %a w%V, %Z', long_format = '%Y-%m-%d %T')
%%>if iswork; then
with open('/proc/cpuinfo', 'r') as f:
	isvm = f.readlines()
isvm = [line.replace('\n', ' ').replace('\t', ' ').split(':') for line in isvm if line]
isvm = any('hypervisor' in ':'.join(line[1:]).split(' ') for line in isvm if line[0].strip().startswith('flags'))
card1 = 'Ensoniq AudioPCI' if isvm else 'HDA Intel PCH'
%%>elif test $hostname = hades; then
card1 = 'HDA Intel PCH'
%%>else
card1 = 'HD-Audio Generic'
%%>fi
mixers1   = ['Master', 'PCM'] ## TODO
%%>if iswork; then
#mixers1.append(('Headphone', 'Speaker'))
if not isvm:
	mixers1.append(MixerAction('Headphone', lambda : spawn_read('headphones')))
	mixers1.append(MixerAction('Speaker', lambda : spawn_read('speakers')))
%%>fi
card2 = 'Yeti Stereo Microphone'
mixers2 = ['Speaker', 'Mic']
#myii         = ...
#myirc        = ...
#from plugins.ii import II
#from myii       import MyII
#def irc_watch(s):
#    s = s.lower()
#    return any(map(lambda x : x in s, ['mattias andrée', 'maandree', 'mandree', 'maandre', 'mandre']))
#ii           = II('irc.oftc.net/#suckless', prefix = HOME + '/.irc')
#myii         = MyII(None, channel = 'irc.oftc.net/#suckless', prefix = HOME + '/.irc', watch = irc_watch)
#myirc        = MyIRC(None, ii = myii)
#myii.display = myirc
mylid        = ...
mybattery    = ...
%%>if test -d /proc/acpi/button/lid; then
mylid        = MyLid    (None)
%%>fi
%%>if test ! $(ls /sys/class/power_supply/ | wc -l) = 0; then
mybattery    = MyBattery(None)
%%>fi

pingthese = []
%%>if test -r ~/.dotfiles/.secrets/ping-"$hostname"; then
%%>    for address in $(cat ~/.dotfiles/.secrets/ping-"$hostname"); do
pingthese.append(Ping(targets = Ping.get_nics('%%{address}'), interval = 30))
%%>    done
%%>fi

try:
    with open(HOME + '/.config/metar', 'rb') as file:
        metar_stations = file.read().decode('utf-8', 'strict').split('\n')
except:
    raise
    try:
        with open('/etc/metar', 'rb') as file:
            metar_stations = file.read().decode('utf-8', 'strict').split('\n')
    except:
            metar_stations = []
metar_stations = [x[0].upper() + x[1:].lower() for x in metar_stations if x != '']

netrenamemap = {
    'lo'              : None,
    'veth42d1872'     : None,
    'enxc84bd6ba1a73' : None,
    'enxc84bd6ba1a91' : None,
    'enx00249b1e3c30' : 'Ctrl',
    'ens33'           : 'Nat',  # VMWare
    'ens37'           : 'Ctrl', # VMWare
    'enxc84d44213db2' : 'Home', # USB-C eth+usb dongle
    'enx0050b6cbd51b' : 'Home', # USB-C eth dongle
    'enxb44506e09918' : 'Eth',
    'wlp0s20f3'       : 'WiFi'
}
netignorelist = [k for k, v in netrenamemap.items() if v is None]
netrenamemap = {k: v for k, v in netrenamemap.items() if v is not None}

functions = [ [ myxmonad
              , None
              , MyTimer     (None, alarms = [])
              , MyALSA      (None, cards = card1, mixers = mixers1, colours = {'Speaker' : '31'})
%%>if test "$hostname" = zenith; then
              , MyALSA      (None, cards = card2, mixers = mixers2, prefix = 'Y.')
%%>fi
              , MyComputer  (lambda f : Clocked(f, 20))
              , myscroll
              , None
              , myclock
              , MyCPU       (lambda f : Clocked(f, 2))
              , MyMemory    (lambda f : Clocked(f, 2))
              , None
              , MyNetwork   (lambda f : Clocked(f, 2), ignore = netignorelist, pings = pingthese, renamemap = netrenamemap)
              #, myirc
%%>if test -x /usr/bin/featherweight; then
              , MyNews      (None)
%%>fi
              , MyWeather   (None, stations = metar_stations) if metar_stations else ...
              #, MySun       (None, clock = myclock)
              ]
            , [ myxmonad
              , None
%%>if test "$hostname" = zenith; then
              , MyIPAddress (lambda f : Clocked(f, 20), public = False)
%%>else
              , MyALSA      (None, cards = card2, mixers = mixers2, prefix = 'Yeti.')
%%>fi
              , MyMOC       (None)
              , myscroll
              , None
              , myclock
              , mylid
              , mybattery
              #, myii
              , None
              , MyStat      (lambda f : Clocked(f, 10))
%%>if test ! "$hostname" = zenith && test ! "$hostname" = north; then ## TODO how about east and hyperion?
              , MyCGNegative()
%%>fi
              , MyBrilliance(None)
%%>if test ! "$hostname" = zenith; then
              , MyBacklight (None)
%%>fi
              #, MyIO        (lambda f : Clocked(f, 10), fs_ignore = [])
              ]
            ]

functions = [[mon for mon in group if mon != ...] for group in functions]

G.HEIGHT = 0
G.groups = [Group(f) for f in functions]
G.groupi = 0
G.group = G.groups[G.groupi]
G.semaphore = threading.Semaphore()

def update_per_clock():
    if G.semaphore.acquire(blocking = False):
        try:
            for g in G.groups:
                for f in g.functions:
                    if isinstance(f.wrapped, Clocked):
                        f(True)
        finally:
            G.semaphore.release()
        invalidate()


start_ = start
def start():
    start_()
    #if myii is not ...:
    #    myii.start()
    bar.clear()
    get_display().flush()
    xasync(lambda : G.clock.continuous_sync(t(update_per_clock)), name = 'clock')
    xasync(mqueue_wait, name = 'mqueue')

def redraw():
    if G.semaphore.acquire(blocking = False):
        try:
            gr = G.groups[G.groupi]
            for g in G.groups:
                if g is not gr:
                    for f in g.functions:
                        f()
            values = gr.pattern % tuple(f() for f in gr.functions)
            bar.partial_clear(0, bar.width, 10, 0, 2, values)
            bar.draw_coloured_splitted_text(0, bar.width, 10, 0, 2, values)
        finally:
            G.semaphore.release()
        return True
    return False

def unhandled_event(e):
    if isinstance(e, Xlib.protocol.event.ButtonPress):
        y = e.event_y
        x = e.event_x
        row = y // HEIGHT_PER_LINE
        lcol = x // bar.font_width
        rcol = (bar.width - x) // bar.font_width
        button = e.detail
        if button in (FORWARD_BUTTON, BACKWARD_BUTTON):
            G.groupi = (G.groupi + (+1 if button == BACKWARD_BUTTON else -1)) % len(G.groups)
            G.group = G.groups[G.groupi]
            invalidate()
        else:
            for f in G.group.posupdate:
                f.update_position()
            for f in G.group.functions:
                if f.click(row, lcol, rcol, button, x, y):
                    break