| import socket
|
| import struct
|
| import json
|
| import os
|
|
|
|
|
| class SwayIPC:
|
| def __init__(self):
|
| self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
| swaysock = os.getenv("SWAYSOCK")
|
| if not swaysock:
|
| raise RuntimeError("SWAYSOCK is not set. Are you inside a Sway session?")
|
| self.sock.connect(swaysock)
|
|
|
| def send(self, msg_type, payload=""):
|
| payload_bytes = payload.encode("utf-8")
|
| header = struct.pack("=6sII", b"i3-ipc", len(payload_bytes), msg_type)
|
| self.sock.sendall(header + payload_bytes)
|
|
|
| def recv(self):
|
| buffer = b""
|
| while len(buffer) < 14:
|
| chunk = self.sock.recv(14 - len(buffer))
|
| if not chunk:
|
| return None
|
| buffer += chunk
|
|
|
| magic, length, msg_type = struct.unpack("=6sII", buffer[:14])
|
| if magic != b"i3-ipc":
|
| raise ValueError("Invalid IPC magic")
|
|
|
| while len(buffer) < 14 + length:
|
| chunk = self.sock.recv(length + 14 - len(buffer))
|
| if not chunk:
|
| return None
|
| buffer += chunk
|
|
|
| return json.loads(buffer[14 : 14 + length])
|
|
|
| def get_tree(self):
|
| self.send(0x80000004)
|
| return self.recv()
|
|
|
|
|
| def list_views(tree):
|
| views = []
|
|
|
| def traverse(node):
|
| if isinstance(node, dict):
|
| if node.get("type") in ["con", "floating_con"] and node.get("window"):
|
| views.append(
|
| {
|
| "id": node["id"],
|
| "title": node.get("name"),
|
| "app_id": node.get("app_id"),
|
| "pid": node.get("pid"),
|
| }
|
| )
|
| for child in node.get("nodes", []) + node.get("floating_nodes", []):
|
| traverse(child)
|
|
|
| traverse(tree)
|
| return views
|
|
|
|
|
| if __name__ == "__main__":
|
| try:
|
| ipc = SwayIPC()
|
| tree = ipc.get_tree()
|
| views = list_views(tree)
|
|
|
| print("Open Views:")
|
| print("-" * 60)
|
| for view in views:
|
| print(f"ID: {view['id']}")
|
| print(f"Title: {view['title']}")
|
| print(f"App ID: {view['app_id'] or 'Unknown'}")
|
| print(f"PID: {view['pid']}")
|
| print("-" * 60)
|
|
|
| except Exception as e:
|
| print(f"[ERROR] {e}")
|