fix(i3blocks): add more error handling

2 files changed, 85 insertions, 105 deletions

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