dotfiles/config/lemonbar/providers.py

#!/usr/bin/env python3

import datetime
from updaters import *
from display import *
import pulsectl
import psutil
import subprocess
import socket
import ipaddress
import logging
import coloredlogs
import json
import notmuch
import mpd
import random
import math

coloredlogs.install(level="DEBUG", fmt="%(levelname)s %(message)s")
log = logging.getLogger()

# TODO Generator class (for I3WorkspacesProvider, NetworkProvider and later
# PulseaudioProvider and MpdProvider)


def humanSize(num):
    """
    Returns a string of width 3+3
    """
    for unit in ("B  ", "KiB", "MiB", "GiB", "TiB", "PiB", "EiB", "ZiB"):
        if abs(num) < 1000:
            if num >= 10:
                return "{:3d}{}".format(int(num), unit)
            else:
                return "{:.1f}{}".format(num, unit)
        num /= 1024.0
    return "{:d}YiB".format(num)


def randomColor(seed=0):
    random.seed(seed)
    return "#{:02x}{:02x}{:02x}".format(*[random.randint(0, 255) for _ in range(3)])


class TimeProvider(StatefulSection, PeriodicUpdater):

    FORMATS = ["%H:%M", "%m-%d %H:%M:%S", "%a %y-%m-%d %H:%M:%S"]
    NUMBER_STATES = len(FORMATS)
    DEFAULT_STATE = 1

    def fetcher(self):
        now = datetime.datetime.now()
        return now.strftime(self.FORMATS[self.state])

    def __init__(self, theme=None):
        PeriodicUpdater.__init__(self)
        StatefulSection.__init__(self, theme)
        self.changeInterval(1)  # TODO OPTI When state < 1


class AlertLevel(enum.Enum):
    NORMAL = 0
    WARNING = 1
    DANGER = 2


class AlertingSection(StatefulSection):
    # TODO EASE Correct settings for themes
    THEMES = {AlertLevel.NORMAL: 2, AlertLevel.WARNING: 3, AlertLevel.DANGER: 1}
    PERSISTENT = True

    def getLevel(self, quantity):
        if quantity > self.dangerThresold:
            return AlertLevel.DANGER
        elif quantity > self.warningThresold:
            return AlertLevel.WARNING
        else:
            return AlertLevel.NORMAL

    def updateLevel(self, quantity):
        self.level = self.getLevel(quantity)
        self.updateTheme(self.THEMES[self.level])
        if self.level == AlertLevel.NORMAL:
            return
        # TODO Temporary update state

    def __init__(self, theme):
        StatefulSection.__init__(self, theme)
        self.dangerThresold = 0.90
        self.warningThresold = 0.75


class CpuProvider(AlertingSection, PeriodicUpdater):
    NUMBER_STATES = 3
    ICON = ""

    def fetcher(self):
        percent = psutil.cpu_percent(percpu=False)
        self.updateLevel(percent / 100)
        if self.state >= 2:
            percents = psutil.cpu_percent(percpu=True)
            return "".join([Section.ramp(p / 100) for p in percents])
        elif self.state >= 1:
            return Section.ramp(percent / 100)

    def __init__(self, theme=None):
        AlertingSection.__init__(self, theme)
        PeriodicUpdater.__init__(self)
        self.changeInterval(1)


class RamProvider(AlertingSection, PeriodicUpdater):
    """
    Shows free RAM
    """

    NUMBER_STATES = 4
    ICON = ""

    def fetcher(self):
        mem = psutil.virtual_memory()
        freePerc = mem.percent / 100
        self.updateLevel(freePerc)

        if self.state < 1:
            return None

        text = Text(Section.ramp(freePerc))
        if self.state >= 2:
            freeStr = humanSize(mem.total - mem.available)
            text.append(freeStr)
        if self.state >= 3:
            totalStr = humanSize(mem.total)
            text.append("/", totalStr)

        return text

    def __init__(self, theme=None):
        AlertingSection.__init__(self, theme)
        PeriodicUpdater.__init__(self)
        self.changeInterval(1)


class TemperatureProvider(AlertingSection, PeriodicUpdater):
    NUMBER_STATES = 2
    RAMP = ""

    def fetcher(self):
        allTemp = psutil.sensors_temperatures()
        if "coretemp" not in allTemp:
            # TODO Opti Remove interval
            return ""
        temp = allTemp["coretemp"][0]

        self.warningThresold = temp.high
        self.dangerThresold = temp.critical
        self.updateLevel(temp.current)

        self.icon = Section.ramp(temp.current / temp.high, self.RAMP)
        if self.state >= 1:
            return "{:.0f}°C".format(temp.current)

    def __init__(self, theme=None):
        AlertingSection.__init__(self, theme)
        PeriodicUpdater.__init__(self)
        self.changeInterval(5)


class BatteryProvider(AlertingSection, PeriodicUpdater):
    # TODO Support ACPID for events
    NUMBER_STATES = 3
    RAMP = ""

    def fetcher(self):
        bat = psutil.sensors_battery()
        if not bat:
            self.icon = None
            return None

        self.icon = ("" if bat.power_plugged else "") + Section.ramp(
            bat.percent / 100, self.RAMP
        )

        self.updateLevel(1 - bat.percent / 100)

        if self.state < 1:
            return

        t = Text("{:.0f}%".format(bat.percent))

        if self.state < 2:
            return t

        h = int(bat.secsleft / 3600)
        m = int((bat.secsleft - h * 3600) / 60)
        t.append(" ({:d}:{:02d})".format(h, m))
        return t

    def __init__(self, theme=None):
        AlertingSection.__init__(self, theme)
        PeriodicUpdater.__init__(self)
        self.changeInterval(5)


class PulseaudioProvider(StatefulSection, ThreadedUpdater):
    NUMBER_STATES = 3
    DEFAULT_STATE = 1

    def __init__(self, theme=None):
        ThreadedUpdater.__init__(self)
        StatefulSection.__init__(self, theme)
        self.pulseEvents = pulsectl.Pulse("event-handler")

        self.pulseEvents.event_mask_set(pulsectl.PulseEventMaskEnum.sink)
        self.pulseEvents.event_callback_set(self.handleEvent)
        self.start()
        self.refreshData()

    def fetcher(self):
        sinks = []
        with pulsectl.Pulse("list-sinks") as pulse:
            for sink in pulse.sink_list():
                if sink.port_active.name == "analog-output-headphones":
                    icon = ""
                elif sink.port_active.name == "analog-output-speaker":
                    icon = "" if sink.mute else ""
                elif sink.port_active.name == "headset-output":
                    icon = ""
                else:
                    icon = "?"
                vol = pulse.volume_get_all_chans(sink)
                fg = (sink.mute and "#333333") or (vol > 1 and "#FF0000") or None

                t = Text(icon, fg=fg)
                sinks.append(t)

                if self.state < 1:
                    continue

                if self.state < 2:
                    if not sink.mute:
                        ramp = " "
                        while vol >= 0:
                            ramp += self.ramp(vol if vol < 1 else 1)
                            vol -= 1
                        t.append(ramp)
                else:
                    t.append(" {:2.0f}%".format(vol * 100))

        return Text(*sinks)

    def loop(self):
        self.pulseEvents.event_listen()

    def handleEvent(self, ev):
        self.refreshData()


class NetworkProviderSection(StatefulSection, Updater):

    NUMBER_STATES = 5
    DEFAULT_STATE = 1

    def actType(self):
        self.ssid = None
        if self.iface.startswith("eth") or self.iface.startswith("enp"):
            if "u" in self.iface:
                self.icon = ""
            else:
                self.icon = ""
        elif self.iface.startswith("wlan") or self.iface.startswith("wl"):
            self.icon = ""
            if self.showSsid:
                cmd = ["iwgetid", self.iface, "--raw"]
                p = subprocess.run(cmd, stdout=subprocess.PIPE)
                self.ssid = p.stdout.strip().decode()
        elif self.iface.startswith("tun") or self.iface.startswith("tap"):
            self.icon = ""
        elif self.iface.startswith("docker"):
            self.icon = ""
        elif self.iface.startswith("veth"):
            self.icon = ""
        elif self.iface.startswith("vboxnet"):
            self.icon = ""
        else:
            self.icon = "?"

    def getAddresses(self):
        ipv4 = None
        ipv6 = None
        for address in self.parent.addrs[self.iface]:
            if address.family == socket.AF_INET:
                ipv4 = address
            elif address.family == socket.AF_INET6:
                ipv6 = address
        return ipv4, ipv6

    def fetcher(self):
        self.icon = None
        self.persistent = False
        if (
            self.iface not in self.parent.stats
            or not self.parent.stats[self.iface].isup
            or self.iface.startswith("lo")
        ):
            return None

        # Get addresses
        ipv4, ipv6 = self.getAddresses()
        if ipv4 is None and ipv6 is None:
            return None

        text = []
        self.persistent = True
        self.actType()

        if self.showSsid and self.ssid:
            text.append(self.ssid)

        if self.showAddress:
            if ipv4:
                netStrFull = "{}/{}".format(ipv4.address, ipv4.netmask)
                addr = ipaddress.IPv4Network(netStrFull, strict=False)
                addrStr = "{}/{}".format(ipv4.address, addr.prefixlen)
                text.append(addrStr)
            # TODO IPV6
            # if ipv6:
            #     text += ' ' + ipv6.address

        if self.showSpeed:
            recvDiff = (
                self.parent.IO[self.iface].bytes_recv
                - self.parent.prevIO[self.iface].bytes_recv
            )
            sentDiff = (
                self.parent.IO[self.iface].bytes_sent
                - self.parent.prevIO[self.iface].bytes_sent
            )
            recvDiff /= self.parent.dt
            sentDiff /= self.parent.dt
            text.append("↓{}↑{}".format(humanSize(recvDiff), humanSize(sentDiff)))

        if self.showTransfer:
            text.append(
                "⇓{}⇑{}".format(
                    humanSize(self.parent.IO[self.iface].bytes_recv),
                    humanSize(self.parent.IO[self.iface].bytes_sent),
                )
            )

        return " ".join(text)

    def onChangeState(self, state):
        self.showSsid = state >= 1
        self.showAddress = state >= 2
        self.showSpeed = state >= 3
        self.showTransfer = state >= 4

    def __init__(self, iface, parent):
        Updater.__init__(self)
        StatefulSection.__init__(self, theme=parent.theme)
        self.iface = iface
        self.parent = parent


class NetworkProvider(Section, PeriodicUpdater):
    def fetchData(self):
        self.prev = self.last
        self.prevIO = self.IO

        self.stats = psutil.net_if_stats()
        self.addrs = psutil.net_if_addrs()
        self.IO = psutil.net_io_counters(pernic=True)
        self.ifaces = self.stats.keys()

        self.last = time.perf_counter()
        self.dt = self.last - self.prev

    def fetcher(self):
        self.fetchData()

        # Add missing sections
        lastSection = self
        for iface in sorted(list(self.ifaces)):
            if iface not in self.sections.keys():
                section = NetworkProviderSection(iface, self)
                lastSection.appendAfter(section)
                self.sections[iface] = section
            else:
                section = self.sections[iface]
            lastSection = section

        # Refresh section text
        for section in self.sections.values():
            section.refreshData()

        return None

    def addParent(self, parent):
        self.parents.add(parent)
        self.refreshData()

    def __init__(self, theme=None):
        PeriodicUpdater.__init__(self)
        Section.__init__(self, theme)

        self.sections = dict()
        self.last = 0
        self.IO = dict()
        self.fetchData()
        self.changeInterval(5)


class RfkillProvider(Section, PeriodicUpdater):
    # TODO FEAT rfkill doesn't seem to indicate that the hardware switch is
    # toggled
    PATH = "/sys/class/rfkill"

    def fetcher(self):
        t = Text()
        for device in os.listdir(self.PATH):
            with open(os.path.join(self.PATH, device, "soft"), "rb") as f:
                softBlocked = f.read().strip() != b"0"
            with open(os.path.join(self.PATH, device, "hard"), "rb") as f:
                hardBlocked = f.read().strip() != b"0"

            if not hardBlocked and not softBlocked:
                continue

            with open(os.path.join(self.PATH, device, "type"), "rb") as f:
                typ = f.read().strip()

            fg = (hardBlocked and "#CCCCCC") or (softBlocked and "#FF0000")
            if typ == b"wlan":
                icon = ""
            elif typ == b"bluetooth":
                icon = ""
            else:
                icon = "?"

            t.append(Text(icon, fg=fg))
        return t

    def __init__(self, theme=None):
        PeriodicUpdater.__init__(self)
        Section.__init__(self, theme)
        self.changeInterval(5)


class SshAgentProvider(PeriodicUpdater):
    def fetcher(self):
        cmd = ["ssh-add", "-l"]
        proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
        if proc.returncode != 0:
            return None
        text = Text()
        for line in proc.stdout.split(b"\n"):
            if not len(line):
                continue
            fingerprint = line.split()[1]
            text.append(Text("", fg=randomColor(seed=fingerprint)))
        return text

    def __init__(self):
        PeriodicUpdater.__init__(self)
        self.changeInterval(5)


class GpgAgentProvider(PeriodicUpdater):
    def fetcher(self):
        cmd = ["gpg-connect-agent", "keyinfo --list", "/bye"]
        proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
        # proc = subprocess.run(cmd)
        if proc.returncode != 0:
            return None
        text = Text()
        for line in proc.stdout.split(b"\n"):
            if not len(line) or line == b"OK":
                continue
            spli = line.split()
            if spli[6] != b"1":
                continue
            keygrip = spli[2]
            text.append(Text("", fg=randomColor(seed=keygrip)))
        return text

    def __init__(self):
        PeriodicUpdater.__init__(self)
        self.changeInterval(5)


class KeystoreProvider(Section, MergedUpdater):
    # TODO OPTI+FEAT Use ColorCountsSection and not MergedUpdater, this is useless
    ICON = ""

    def __init__(self, theme=None):
        MergedUpdater.__init__(self, SshAgentProvider(), GpgAgentProvider())
        Section.__init__(self, theme)


class NotmuchUnreadProvider(ColorCountsSection, InotifyUpdater):
    COLORABLE_ICON = ""

    def subfetcher(self):
        db = notmuch.Database(mode=notmuch.Database.MODE.READ_ONLY, path=self.dir)
        counts = []
        for account in self.accounts:
            queryStr = "folder:/{}/ and tag:unread".format(account)
            query = notmuch.Query(db, queryStr)
            nbMsgs = query.count_messages()
            if account == "frogeye":
                global q
                q = query
            if nbMsgs < 1:
                continue
            counts.append((nbMsgs, self.colors[account]))
        # db.close()
        return counts

    def __init__(self, dir="~/.mail/", theme=None):
        PeriodicUpdater.__init__(self)
        ColorCountsSection.__init__(self, theme)

        self.dir = os.path.realpath(os.path.expanduser(dir))
        assert os.path.isdir(self.dir)

        # Fetching account list
        self.accounts = sorted(
            [a for a in os.listdir(self.dir) if not a.startswith(".")]
        )
        # Fetching colors
        self.colors = dict()
        for account in self.accounts:
            filename = os.path.join(self.dir, account, "color")
            with open(filename, "r") as f:
                color = f.read().strip()
            self.colors[account] = color

        self.addPath(os.path.join(self.dir, ".notmuch", "xapian"))


class TodoProvider(ColorCountsSection, InotifyUpdater):
    # TODO OPT/UX Maybe we could get more data from the todoman python module
    # TODO OPT Specific callback for specific directory
    COLORABLE_ICON = ""

    def updateCalendarList(self):
        calendars = sorted(os.listdir(self.dir))
        for calendar in calendars:
            # If the calendar wasn't in the list
            if calendar not in self.calendars:
                self.addPath(os.path.join(self.dir, calendar), refresh=False)

                # Fetching name
                path = os.path.join(self.dir, calendar, "displayname")
                with open(path, "r") as f:
                    self.names[calendar] = f.read().strip()

                # Fetching color
                path = os.path.join(self.dir, calendar, "color")
                with open(path, "r") as f:
                    self.colors[calendar] = f.read().strip()
        self.calendars = calendars

    def __init__(self, dir, theme=None):
        """
        :parm str dir: [main]path value in todoman.conf
        """
        InotifyUpdater.__init__(self)
        ColorCountsSection.__init__(self, theme=theme)
        self.dir = os.path.realpath(os.path.expanduser(dir))
        assert os.path.isdir(self.dir)

        self.calendars = []
        self.colors = dict()
        self.names = dict()
        self.updateCalendarList()
        self.refreshData()

    def countUndone(self, calendar):
        cmd = ["todo", "--porcelain", "list"]
        if calendar:
            cmd.append(self.names[calendar])
        proc = subprocess.run(cmd, stdout=subprocess.PIPE)
        data = json.loads(proc.stdout)
        return len(data)

    def subfetcher(self):
        counts = []

        # TODO This an ugly optimisation that cuts on features, but todoman
        # calls are very expensive so we keep that in the meanwhile
        if self.state < 2:
            c = self.countUndone(None)
            if c > 0:
                counts.append((c, "#00000"))
                counts.append((0, "#FFFFF"))
            return counts
        # Optimisation ends here

        for calendar in self.calendars:
            c = self.countUndone(calendar)
            if c <= 0:
                continue
            counts.append((c, self.colors[calendar]))
        return counts


class I3WindowTitleProvider(Section, I3Updater):
    # TODO FEAT To make this available from start, we need to find the
    # `focused=True` element following the `focus` array
    # TODO Feat Make this output dependant if wanted
    def on_window(self, i3, e):
        self.updateText(e.container.name)

    def __init__(self, theme=None):
        I3Updater.__init__(self)
        Section.__init__(self, theme=theme)
        self.on("window", self.on_window)


class I3WorkspacesProviderSection(Section):
    def selectTheme(self):
        if self.urgent:
            return self.parent.themeUrgent
        elif self.focused:
            return self.parent.themeFocus
        else:
            return self.parent.themeNormal

    # TODO On mode change the state (shown / hidden) gets overriden so every
    # tab is shown

    def show(self):
        self.updateTheme(self.selectTheme())
        self.updateText(self.fullName if self.focused else self.shortName)

    def changeState(self, focused, urgent):
        self.focused = focused
        self.urgent = urgent
        self.show()

    def setName(self, name):
        self.shortName = name
        self.fullName = (
            self.parent.customNames[name] if name in self.parent.customNames else name
        )

    def switchTo(self):
        self.parent.i3.command("workspace {}".format(self.shortName))

    def __init__(self, name, parent):
        Section.__init__(self)
        self.parent = parent
        self.setName(name)
        self.setDecorators(clickLeft=self.switchTo)
        self.tempText = None

    def empty(self):
        self.updateTheme(self.parent.themeNormal)
        self.updateText(None)

    def tempShow(self):
        self.updateText(self.tempText)

    def tempEmpty(self):
        self.tempText = self.dstText[1]
        self.updateText(None)


class I3WorkspacesProvider(Section, I3Updater):
    # TODO FEAT Multi-screen

    def initialPopulation(self, parent):
        """
        Called on init
        Can't reuse addWorkspace since i3.get_workspaces() gives dict and not
        ConObjects
        """
        workspaces = self.i3.get_workspaces()
        lastSection = self.modeSection
        for workspace in workspaces:
            # if parent.display != workspace["display"]:
            #     continue

            section = I3WorkspacesProviderSection(workspace.name, self)
            section.focused = workspace.focused
            section.urgent = workspace.urgent
            section.show()
            parent.addSectionAfter(lastSection, section)
            self.sections[workspace.num] = section

            lastSection = section

    def on_workspace_init(self, i3, e):
        workspace = e.current
        i = workspace.num
        if i in self.sections:
            section = self.sections[i]
        else:
            # Find the section just before
            while i not in self.sections.keys() and i > 0:
                i -= 1
            prevSection = self.sections[i] if i != 0 else self.modeSection

            section = I3WorkspacesProviderSection(workspace.name, self)
            prevSection.appendAfter(section)
            self.sections[workspace.num] = section
        section.focused = workspace.focused
        section.urgent = workspace.urgent
        section.show()

    def on_workspace_empty(self, i3, e):
        self.sections[e.current.num].empty()

    def on_workspace_focus(self, i3, e):
        self.sections[e.old.num].focused = False
        self.sections[e.old.num].show()
        self.sections[e.current.num].focused = True
        self.sections[e.current.num].show()

    def on_workspace_urgent(self, i3, e):
        self.sections[e.current.num].urgent = e.current.urgent
        self.sections[e.current.num].show()

    def on_workspace_rename(self, i3, e):
        self.sections[e.current.num].setName(e.name)
        self.sections[e.current.num].show()

    def on_mode(self, i3, e):
        if e.change == "default":
            self.modeSection.updateText(None)
            for section in self.sections.values():
                section.tempShow()
        else:
            self.modeSection.updateText(e.change)
            for section in self.sections.values():
                section.tempEmpty()

    def __init__(
        self, theme=0, themeFocus=3, themeUrgent=1, themeMode=2, customNames=dict()
    ):
        I3Updater.__init__(self)
        Section.__init__(self)
        self.themeNormal = theme
        self.themeFocus = themeFocus
        self.themeUrgent = themeUrgent
        self.customNames = customNames

        self.sections = dict()
        self.on("workspace::init", self.on_workspace_init)
        self.on("workspace::focus", self.on_workspace_focus)
        self.on("workspace::empty", self.on_workspace_empty)
        self.on("workspace::urgent", self.on_workspace_urgent)
        self.on("workspace::rename", self.on_workspace_rename)
        # TODO Un-handled/tested: reload, rename, restored, move

        self.on("mode", self.on_mode)
        self.modeSection = Section(theme=themeMode)

    def addParent(self, parent):
        self.parents.add(parent)
        parent.addSection(self.modeSection)
        self.initialPopulation(parent)


class MpdProvider(Section, ThreadedUpdater):
    # TODO FEAT More informations and controls

    MAX_LENGTH = 50

    def connect(self):
        self.mpd.connect("localhost", 6600)

    def __init__(self, theme=None):
        ThreadedUpdater.__init__(self)
        Section.__init__(self, theme)

        self.mpd = mpd.MPDClient()
        self.connect()
        self.refreshData()
        self.start()

    def fetcher(self):
        stat = self.mpd.status()
        if not len(stat) or stat["state"] == "stop":
            return None

        cur = self.mpd.currentsong()
        if not len(cur):
            return None

        infos = []

        def tryAdd(field):
            if field in cur:
                infos.append(cur[field])

        tryAdd("title")
        tryAdd("album")
        tryAdd("artist")

        infosStr = " - ".join(infos)
        if len(infosStr) > MpdProvider.MAX_LENGTH:
            infosStr = infosStr[: MpdProvider.MAX_LENGTH - 1] + "…"

        return " {}".format(infosStr)

    def loop(self):
        try:
            self.mpd.idle("player")
            self.refreshData()
        except mpd.base.ConnectionError as e:
            log.warn(e, exc_info=True)
            self.connect()
        except BaseException as e:
            log.error(e, exc_info=True)