Various improvements

notes

@@ -0,0 +1,18 @@
+output DP-3 Dell Inc. DELL U3219Q 9JKK413 1.0 270
+output HDMI-A-1 Sony SONY TV 0x00000101 1.0 normal
+output eDP-1 Unknown 0x1403 0x00000000 None None
+
+tags = [
+    "",
+    "master-of-seat",
+    "mutter-device-disable-kms-modifiers",
+    "seat",
+    "snap_slack_slack",
+    "uaccess",
+]
+
+Watcher is generating too many events. Should consider the state of the laptop lid. If the lid is closed that is significant.
+
+Use MonitorObserver to monitor for events in a background thread.
+
+There can be a short delay before acting on an event.

toggle.py

@@ -0,0 +1,63 @@
+#!/usr/bin/python3
+
+import json
+import os
+import subprocess
+import time
+import typing
+
+
+def run_swaymsg() -> bytes:
+    """Run swaymsg to get list of outputs."""
+    p = subprocess.run(["swaymsg", "-r", "-t", "get_outputs"], capture_output=True)
+    return p.stdout
+
+
+def parse_json(outputs: bytes) -> dict[str, dict[str, typing.Any]]:
+    """Parse get_outputs JSON."""
+    try:
+        json_data = json.loads(outputs)
+    except json.decoder.JSONDecodeError:
+        print("JSON parse error")
+        print(outputs)
+        raise
+
+    return {o["name"]: o for o in json_data}
+
+
+def sony_tv_connected() -> None:
+    """Run command when HDMI is connected."""
+    subprocess.run([os.path.expanduser("~/bin/desk")])
+
+
+def hdmi_disconnected() -> None:
+    """Run command when HDMI is disconnected."""
+    subprocess.run([os.path.expanduser("~/bin/unplugged")])
+
+
+def get_outputs(attempts: int = 10) -> dict[str, dict[str, typing.Any]] | None:
+    """Ask sway for the current list of outputs."""
+    for attempt in range(attempts):
+        outputs = run_swaymsg()
+        if outputs.strip() == b"":
+            return None
+        try:
+            return parse_json(outputs)
+        except json.decoder.JSONDecodeError:
+            if attempt == attempts - 1:
+                raise
+            time.sleep(1)
+    return None
+
+
+def main() -> None:
+    outputs = get_outputs()
+    assert outputs
+    if outputs["eDP-1"]["active"]:
+        sony_tv_connected()
+    else:
+        hdmi_disconnected()
+
+
+if __name__ == "__main__":
+    main()

watch.py

@@ -29,6 +29,18 @@ def run_swaymsg() -> bytes:
     return p.stdout
 
 
+def parse_json(outputs: bytes) -> dict[str, dict[str, Any]]:
+    """Parse get_outputs JSON."""
+    try:
+        json_data = json.loads(outputs)
+    except json.decoder.JSONDecodeError:
+        print("JSON parse error")
+        print(outputs)
+        raise
+
+    return {o["name"]: o for o in json_data}
+
+
 def get_outputs(attempts: int = 10) -> dict[str, dict[str, Any]] | None:
     """Ask sway for the current list of outputs."""
     for attempt in range(attempts):
@@ -36,16 +48,12 @@ def get_outputs(attempts: int = 10) -> dict[str, dict[str, Any]] | None:
         if outputs.strip() == b"":
             return None
         try:
-            json_data = json.loads(outputs)
-            break
+            return parse_json(outputs)
         except json.decoder.JSONDecodeError:
-            print("JSON parse error")
-            print(outputs)
             if attempt == attempts - 1:
                 raise
             time.sleep(1)
-
-    return {o["name"]: o for o in json_data}
+    return None
 
 
 def is_sony_tv(output: dict[str, Any]) -> bool:
@@ -80,7 +88,9 @@ def handle_state_change() -> str:
         hdmi_disconnected()
         return "HDMI disconnected, switching to laptop screen."
 
-    hdmi = get_outputs().get(hdmi_name)
+    outputs = get_outputs()
+
+    hdmi = outputs and outputs.get(hdmi_name)
 
     if hdmi and is_sony_tv(hdmi):
         sony_tv_connected()