From 057b55d60fc11e8f591ea2792543e71082734c84 Mon Sep 17 00:00:00 2001 From: Toby Vincent Date: Wed, 5 Jun 2024 19:40:16 -0500 Subject: fix(i3blocks): add more error handling --- i3blocks/.local/lib/i3blocks/i3blocks-miniflux | 28 +++-- i3blocks/.local/lib/i3blocks/i3blocks-volume | 162 ++++++++++--------------- 2 files changed, 85 insertions(+), 105 deletions(-) (limited to 'i3blocks') diff --git a/i3blocks/.local/lib/i3blocks/i3blocks-miniflux b/i3blocks/.local/lib/i3blocks/i3blocks-miniflux index 60175e1..f87f1d3 100755 --- a/i3blocks/.local/lib/i3blocks/i3blocks-miniflux +++ b/i3blocks/.local/lib/i3blocks/i3blocks-miniflux @@ -8,23 +8,31 @@ import requests ICON = "\U000f046b" # 󰑫 +def get_count(): + url = "https://miniflux.tobyvin.dev/v1/feeds/counters" + headers = {"X-Auth-Token": os.environ.get("MINIFLUX_TOKEN")} + resp = requests.get(url, headers=headers) + return sum(resp.json().get("unreads").values()) + + def main(): match int(os.environ.get("BLOCK_BUTTON", "0")): case 1: subprocess.run(["xdg-open", "https://miniflux.tobyvin.dev"]) - url = "https://miniflux.tobyvin.dev/v1/feeds/counters" - headers = {"X-Auth-Token": os.environ.get("MINIFLUX_TOKEN")} - resp = requests.get(url, headers=headers) - count = sum(resp.json().get("unreads").values()) + try: + count = get_count() + except requests.exceptions.ConnectionError: + count = "x" print(f" {ICON} {count} \n") - if count > 10: - print(f"#{os.environ.get("BASE16_COLOR_00_HEX")}") - print(f"#{os.environ.get("BASE16_COLOR_08_HEX")}") - elif count > 0: - print(f"#{os.environ.get("BASE16_COLOR_00_HEX")}") - print(f"#{os.environ.get("BASE16_COLOR_0A_HEX")}") + match count: + case str(c) | int(c) if c > 10: + print(f"#{os.environ.get("BASE16_COLOR_00_HEX")}") + print(f"#{os.environ.get("BASE16_COLOR_08_HEX")}") + case int(c) if c > 0: + print(f"#{os.environ.get("BASE16_COLOR_00_HEX")}") + print(f"#{os.environ.get("BASE16_COLOR_0A_HEX")}") if __name__ == "__main__": diff --git a/i3blocks/.local/lib/i3blocks/i3blocks-volume b/i3blocks/.local/lib/i3blocks/i3blocks-volume index c9d6e54..4733add 100755 --- a/i3blocks/.local/lib/i3blocks/i3blocks-volume +++ b/i3blocks/.local/lib/i3blocks/i3blocks-volume @@ -7,105 +7,77 @@ import subprocess import sys -class PulseAudio: - icons = { - "mute": "\U000f075f", # 󰝟 - "low": "\U000f057f", # 󰕿 - "medium": "\U000f0580", # 󰖀 - "high": "\U000f057e", # 󰕾 - } - - def __init__(self): - self.print_status() - - @property - def volume(self) -> dict: - channels = self.sink.get("volume") - - return int( - channels.get(next(chan for chan in channels)) - .get("value_percent") - .rstrip("%") - ) - - @property - def icon(self) -> str: - if self.sink.get("mute"): - return self.icons["mute"] - - match self.volume: - case v if v < 33: - return self.icons["low"] - case v if v < 66: - return self.icons["medium"] - case _: - return self.icons["high"] - - def list_sinks() -> dict[int, dict]: - return json.loads( - subprocess.run( - ["pactl", "--format=json", "list", "sinks"], - capture_output=True, - encoding="UTF-8", - ).stdout - ) - - def get_default_sink() -> str: - return subprocess.run( - ["pactl", "get-default-sink"], +ICONS = { + "mute": "\U000f075f", # 󰝟 + "low": "\U000f057f", # 󰕿 + "medium": "\U000f0580", # 󰖀 + "high": "\U000f057e", # 󰕾 +} + + +def is_muted(): + return ( + subprocess.run( + ["pactl", "get-sink-mute", "@DEFAULT_SINK@"], capture_output=True, encoding="UTF-8", - ).stdout - - def print_status(self, update=True, format="json"): - if update: - self.sinks = PulseAudio.list_sinks() - self.default_sink = PulseAudio.get_default_sink() - self.sink = next(sink for sink in self.sinks) - - output = {"full_text": f" {self.icon} {self.volume}% "} - - if self.sink.get("mute"): - output["color"] = f"#{os.environ.get("BASE16_COLOR_00_HEX")}" - output["background"] = f"#{os.environ.get("BASE16_COLOR_0A_HEX")}" - - match format: - case "json": - print(json.dumps(output, ensure_ascii=False), flush=True) - case "raw": - for prop in ["full_text", "short_text", "color", "background"]: - print(output.get(prop, ""), flush=True) - case f: - raise ValueError(f"Invalid output format: {f}") - - async def listener(self): - process = await asyncio.create_subprocess_exec( - "pactl", - "--format=json", - "subscribe", - stdout=asyncio.subprocess.PIPE, ) + .stdout.removeprefix("Mute: ") + .strip() + == "yes" + ) + + +def get_volume(): + stdout = subprocess.run( + ["pactl", "get-sink-volume", "@DEFAULT_SINK@"], + capture_output=True, + encoding="UTF-8", + ).stdout.strip() + + for s in stdout.removeprefix("Volume: ").split(): + if s.endswith("%"): + return int(s.rstrip("%")) + + +def print_status(): + match get_volume(): + case None: + print(flush=True) + case v if is_muted(): + output = { + "full_text": f" {ICONS["mute"]} {v}% ", + "color": f"#{os.environ.get("BASE16_COLOR_00_HEX")}", + "background": f"#{os.environ.get("BASE16_COLOR_0A_HEX")}", + } + case v if v > 66: + output = {"full_text": f" {ICONS["high"]} {v}% "} + case v if v > 33: + output = {"full_text": f" {ICONS["medium"]} {v}% "} + case _: + output = {"full_text": f" {ICONS["low"]} {v}% "} + + print(json.dumps(output, ensure_ascii=False), flush=True) + + +async def listener(): + process = await asyncio.create_subprocess_exec( + "pactl", + "--format=json", + "subscribe", + stdout=asyncio.subprocess.PIPE, + ) - while True: - line = await process.stdout.readline() + while True: + line = await process.stdout.readline() - if not line: - await asyncio.sleep(1) - continue + if not line: + await asyncio.sleep(1) + continue - match json.loads(line.decode("UTF-8")): - case { - "index": index, - "event": "change", - "on": "sink", - } if index == self.sink.get("index"): - self.print_status() - case { - "index": index, - "event": "change", - "on": "source-output", - }: - self.print_status() + match json.loads(line.decode("UTF-8")): + case {"on": "sink"} | {"on": "source-output"}: + print_status() async def button_handler(): @@ -143,10 +115,10 @@ async def button_handler(): async def main(): - pulse_audio = PulseAudio() + print_status() try: async with asyncio.TaskGroup() as task_group: - task_group.create_task(pulse_audio.listener()) + task_group.create_task(listener()) task_group.create_task(button_handler()) except asyncio.CancelledError: return -- cgit v1.2.3-70-g09d2