feat: Refactor to use brotab Python client directly (#5)

* Changed logic of subprocessing

This fork is for everyone as me that have trouble with subprocess.run, that brotab was not accepting any arguments, son i decided to implement the same functionality importing the brotab package and just calling the respective functions.

* Cleaned prints for bad debugging

* Added function to close tabs.

Created option to close tabs with keyboard cltab, cl or clt, due to limitations on my undertanding of how to manage it it is limited for that words.

brotab_ulauncher/brotab.py

@@ -0,0 +1,67 @@
+import logging
+from string import ascii_lowercase
+
+from typing import Tuple, List
+
+from brotab.inout import is_port_accepting_connections
+from brotab.inout import get_mediator_ports
+from brotab.inout import in_temp_dir
+from brotab.api import SingleMediatorAPI, MultipleMediatorsAPI
+import re
+
+FORMAT = '%(asctime)-15s %(levelname)-10s %(message)s'
+logging.basicConfig(
+    format=FORMAT,
+    filename=in_temp_dir('brotab.log'),
+    level=logging.DEBUG)
+logger = logging.getLogger('brotab')
+logger.info('Logger has been created')
+
+
+def parse_target_hosts(target_hosts: str) -> Tuple[List[str], List[int]]:
+    """
+    Input: localhost:2000,127.0.0.1:3000
+    Output: (['localhost', '127.0.0.1'], [2000, 3000])
+    """
+    hosts, ports = [], []
+    for pair in target_hosts.split(','):
+        host, port = pair.split(':')
+        hosts.append(host)
+        ports.append(int(port))
+    return hosts, ports
+
+
+def create_clients(target_hosts=None) -> List[SingleMediatorAPI]:
+    if target_hosts is None:
+        ports = list(get_mediator_ports())
+        hosts = ['localhost'] * len(ports)
+    else:
+        hosts, ports = parse_target_hosts(target_hosts)
+
+    result = [SingleMediatorAPI(prefix, host=host, port=port)
+              for prefix, host, port in zip(ascii_lowercase, hosts, ports)
+              if is_port_accepting_connections(port, host)]
+    logger.info('Created clients: %s', result)
+    return result
+
+
+def return_tabs():
+    logger.info('Listing tabs')
+    api = MultipleMediatorsAPI(create_clients())
+    tabs = api.list_tabs([])
+    return tabs
+
+
+def activate_tab(prefix):
+    api = MultipleMediatorsAPI(create_clients())
+    api.activate_tab([prefix], True)
+
+def close_tab(prefix):
+    # Try stdin if arguments are empty
+    logger.info('Closing tabs: %s', prefix)
+    api = MultipleMediatorsAPI(create_clients())
+    tabs = api.close_tabs([prefix])
+
+def return_clients():
+    logger.info('Showing clients')
+    return create_clients()

brotab/client.py → brotab_ulauncher/client.py

@@ -1,6 +1,7 @@
 import subprocess
 import logging
-
+from brotab_ulauncher.brotab import return_clients, return_tabs, activate_tab, close_tab
+from asyncio import new_event_loop, set_event_loop
 logger = logging.getLogger(__name__)
 
 
@@ -33,27 +34,20 @@ def index(self):
     def index_clients(self):
         """ Index the clients connected """
         self.clients = {}
-
-        result = subprocess.run(['brotab', 'clients'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
-
-        output_text = result.stdout.decode("utf-8")
-        for line in output_text.splitlines():
-            client_info = line.rstrip().split("\t")
-            self.clients[client_info[0][:1]] = client_info[3]
+        clients = return_clients()
+        for client in clients:
+            self.clients[client.__dict__["_prefix"].replace(".", "")] = client.__dict__[
+                "_browser"]
 
     def index_tabs(self):
         """ Index Tabs list """
-
         self.tabs = []
+        loop = new_event_loop()
+        set_event_loop(loop)
+        tabs_listed = return_tabs()
 
-        result = subprocess.run(['brotab', 'list'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
-
-        output_text = result.stdout.decode("utf-8")
-        if not output_text.strip():
-            return
-
-        for line in output_text.splitlines():
-            tab = line.rstrip().split("\t")
+        for tab in tabs_listed:
+            tab = tab.split("\t")
             self.tabs.append({
                 "prefix": tab[0],
                 "name": tab[1],
@@ -76,7 +70,11 @@ def search_tabs(self, filter_term=None):
 
     def activate_tab(self, prefix):
         """ Activates the tab with the specified prefix """
-        subprocess.run(['brotab', 'activate', prefix, '--focused'])
+        activate_tab(prefix)
+
+    def close_tab(self, prefix):
+        """ Closes the tab with the specified prefix """
+        close_tab(prefix)
 
     def get_browser_icon_from_prefix(self, prefix):
         """ Returns the name of the icon to display as client """

brotab/extension.py → brotab_ulauncher/extension.py

@@ -10,9 +10,11 @@
 from ulauncher.api.shared.action.CopyToClipboardAction import CopyToClipboardAction
 from ulauncher.api.shared.action.HideWindowAction import HideWindowAction
 from ulauncher.api.shared.action.ExtensionCustomAction import ExtensionCustomAction
-from brotab.client import BrotabClient
-from brotab.listeners import KeywordQueryEventListener, ItemEnterEventListener
-from brotab.actions import RESULT_ITEM_ENTER, REFRESH_TABS
+from brotab_ulauncher.client import BrotabClient
+from brotab_ulauncher.listeners import KeywordQueryEventListener, ItemEnterEventListener
+from brotab_ulauncher.actions import RESULT_ITEM_ENTER, REFRESH_TABS
+import gi
+gi.require_version('Notify', '0.7')
 from gi.repository import Notify
 
 DISPLAY_MAX_RESULTS = 20
@@ -26,11 +28,10 @@ def __init__(self):
         super(BrotabExtension, self).__init__()
 
         self.logger.info("Initializing Brotab Extension")
-
         Notify.init(__name__)
 
         self.brotab_client = BrotabClient()
-
+        self.mode = "activator"
         if not self.brotab_client.is_installed():
             raise EnvironmentError("Brotab is not installed on your system. \
                     Please see https://github.com/balta2ar/brotab for instructions.")
@@ -42,8 +43,8 @@ def __init__(self):
 
     def index_tabs(self):
         """ Index brotab tabs """
+        self.brotab_client = BrotabClient()
         self.brotab_client.index()
-
         Timer(INDEX_REFRESH_TIME_SECONDS, self.index_tabs).start()
 
     def show_commands(self, arg):
@@ -76,7 +77,7 @@ def search_tabs(self, event):
         tabs = self.brotab_client.search_tabs(event.get_argument())
 
         for tab in tabs[:DISPLAY_MAX_RESULTS]:
-            data = {"action": RESULT_ITEM_ENTER, 'tab': tab['prefix']}
+            data = {"action": RESULT_ITEM_ENTER, 'tab': tab['prefix'], 'mode': self.mode}
 
             items.append(
                 ExtensionSmallResultItem(icon='images/%s' % tab["icon"],

brotab/listeners.py → brotab_ulauncher/listeners.py

@@ -1,17 +1,24 @@
 from ulauncher.api.client.EventListener import EventListener
-from brotab.actions import RESULT_ITEM_ENTER, REFRESH_TABS
+from brotab_ulauncher.actions import RESULT_ITEM_ENTER, REFRESH_TABS
 from ulauncher.api.shared.action.SetUserQueryAction import SetUserQueryAction
 
 
 class KeywordQueryEventListener(EventListener):
     """ Listener that handles the user input """
+
     def on_event(self, event, extension):
         """ Handles the event """
+        keyword = event.get_keyword() or ""
+        valid_keywords = ["cltab","cl","clt"]
+        if keyword in valid_keywords:
+            extension.mode = "killer"
+        else:
+            extension.mode = "activator"
+
         argument = event.get_argument() or ""
 
         if argument.startswith(":"):
             return extension.show_commands(argument)
-
         return extension.search_tabs(event)
 
 
@@ -22,12 +29,19 @@ class ItemEnterEventListener(EventListener):
     def on_event(self, event, extension):
         """ Handles the event """
         data = event.get_data()
-
-        if data["action"] == RESULT_ITEM_ENTER:
+        if data["action"] == RESULT_ITEM_ENTER and data["mode"] == "activator":
             extension.brotab_client.activate_tab(data["tab"])
-
+        if data["action"] == RESULT_ITEM_ENTER and data["mode"] == "killer":
+            try:
+                extension.brotab_client.close_tab(data["tab"])
+            except Exception as error:
+                print(error)
+            extension.notify("Tab closed")
+            print("Tab closed")
         if data["action"] == REFRESH_TABS:
             extension.brotab_client.index()
-            kw = extension.preferences["kw"]
+            print(extension.preferences.keys())
+            dict_mode = {"activator": "kw", "killer": "kw_cl"}
+            kw = extension.preferences[dict_mode[extension.mode]]
             extension.notify("Index Finished")
             return SetUserQueryAction("%s " % kw)

main.py

@@ -2,7 +2,7 @@
 Main Extension Entrypoint
 """
 
-from brotab.extension import BrotabExtension
+from brotab_ulauncher.extension import BrotabExtension
 
 if __name__ == '__main__':
     BrotabExtension().run()

manifest.json

@@ -14,6 +14,13 @@
             "name": "Brotab",
             "description": "Control your browsers tabs",
             "default_value": "brotab"
+        },
+        {
+            "id": "kw_cl",
+            "type": "keyword",
+            "name": "Closetab",
+            "description": "Close an active tab WARNING: Only is ready to accept cltab, cl or clt keywords",
+            "default_value": "cltab"
         }
     ]
 }