from i3sway.protocol import Connection, MessageType class Monitor: tracked_classes = None cmd_sock = None subscription_sock = None event_handler = None conn_type = None def __enter__(self): return self def __exit__(self, *details): self.cmd_sock.close() self.subscription_sock.close() def __init__(self, event_handler, tracked_classes): self.event_handler = event_handler self.tracked_classes = tracked_classes self.cmd_sock = Connection() self.subscription_sock = Connection() tree = self.cmd_sock.request(MessageType.GET_TREE).payload def parse_tree(root): if root.get('window') is not None or root.get('app_id') is not None: # Application window for name, classes in tracked_classes.items(): if self.container_matches_classes(root, classes): id = str(root['id']) self.event_handler.on_create(name, id) if root['visible']: self.event_handler.on_visible(name, id) if root['type'] == 'floating_con': self.event_handler.on_float(name, id) elif root['type'] == 'con': self.event_handler.on_tile(name, id) else: for node in root['nodes']: parse_tree(node) for node in root['floating_nodes']: parse_tree(node) parse_tree(tree) def container_matches_classes(self, container, classes): normalize = lambda v: v.lower() if v is not None else '' app_id = container.get('app_id') window_props = container.get('window_properties') instance = window_props.get('instance') if window_props is not None else None window_class = window_props.get('class') if window_props is not None else None app_id = normalize(app_id) instance = normalize(instance) window_class = normalize(window_class) for keyword in classes: kw = normalize(keyword) if kw in app_id or kw in instance or kw in window_class: return True return False def listen(self): for event in self.subscription_sock.subscribe(['window']): self.handle_event(event) def is_window_in_scratchpad(self, id): def scan_tree(root, is_in_scratch): if root['id'] == id: return is_in_scratch is_scratch = is_in_scratch or ( root['type'] == 'workspace' and root['name'] == '__i3_scratch' ) for node in root['nodes'] + root['floating_nodes']: result = scan_tree(node, is_scratch) if result is not None: return result tree = self.cmd_sock.request(MessageType.GET_TREE) return scan_tree(tree.payload, False) def handle_event(self, event): if event.type == MessageType.window: return self.handle_window_event(event) def handle_window_event(self, event): container = event['container'] for name, classes in self.tracked_classes.items(): change = event['change'] if self.container_matches_classes(container, classes): id = str(container['id']) if change == 'new': self.event_handler.on_create(name, id) elif change == 'focus': self.event_handler.on_visible(name, id) elif change == 'close': self.event_handler.on_destroy(name, id) elif change == 'move': if container['visible']: self.event_handler.on_visible(name, id) else: self.event_handler.on_invisible(name, id) elif change == 'floating': if container['type'] == 'floating_con': self.event_handler.on_float(name, id) elif container['type'] == 'con': self.event_handler.on_tile(name, id)