refactor: tidy codes

Signed-off-by: Jack Cherng <[email protected]>

plugin/commands/auto_set_syntax.py

@@ -1,26 +1,25 @@
 from __future__ import annotations
+from collections.abc import Callable
 
 import re
-import uuid
-from contextlib import contextmanager
 from functools import wraps
 from itertools import chain
 from operator import itemgetter
 from pathlib import Path
-from typing import Any, Callable, Generator, Iterable, Mapping, TypeVar, cast
+from typing import Any, Mapping, TypeVar, cast
 
 import sublime
 import sublime_plugin
 
-from ..constants import PLUGIN_NAME, RE_ST_SYNTAX_TEST_LINE, RE_VIM_SYNTAX_LINE, VIEW_RUN_ID_SETTINGS_KEY
+from ..constants import PLUGIN_NAME, RE_ST_SYNTAX_TEST_LINE, RE_VIM_SYNTAX_LINE
 from ..guesslang.types import GuesslangServerPredictionItem, GuesslangServerResponse
 from ..helpers import is_syntaxable_view
 from ..libs import websocket
 from ..logger import Logger
 from ..rules import SyntaxRuleCollection
 from ..settings import get_merged_plugin_setting, get_merged_plugin_settings, pref_trim_suffixes
 from ..shared import G
-from ..snapshot import ViewSnapshot, ViewSnapshotCollection
+from ..snapshot import ViewSnapshot
 from ..types import ListenerEvent
 from ..utils import (
     find_syntax_by_syntax_like,
@@ -140,20 +139,6 @@ def _status_msg_and_log(msg: str, window: sublime.Window | None = None) -> None:
         sublime.status_message(msg)
 
 
-@contextmanager
-def _view_snapshot_context(view: sublime.View) -> Generator[ViewSnapshot, None, None]:
-    run_id = str(uuid.uuid4())
-    settings = view.settings()
-
-    try:
-        settings.set(VIEW_RUN_ID_SETTINGS_KEY, run_id)
-        ViewSnapshotCollection.add(run_id, view)
-        yield ViewSnapshotCollection.get(run_id)  # type: ignore
-    finally:
-        settings.erase(VIEW_RUN_ID_SETTINGS_KEY)
-        ViewSnapshotCollection.pop(run_id)
-
-
 def _snapshot_view(failed_ret: Any = None) -> Callable[[_T_Callable], _T_Callable]:
     def decorator(func: _T_Callable) -> _T_Callable:
         @wraps(func)
@@ -162,7 +147,7 @@ def wrapped(view: sublime.View, *args: Any, **kwargs: Any) -> Any:
                 Logger.log("⏳ Calm down! View has gone or the plugin is not ready yet.")
                 return failed_ret
 
-            with _view_snapshot_context(view):
+            with G.view_snapshot_collection.snapshot_context(view):
                 return func(view, *args, **kwargs)
 
         return cast(_T_Callable, wrapped)
@@ -184,7 +169,7 @@ def run_auto_set_syntax_on_view(
     if not (
         (window := view.window())
         and is_syntaxable_view(view, must_plaintext)
-        and (syntax_rule_collection := G.get_syntax_rule_collection(window))
+        and (syntax_rule_collection := G.syntax_rule_collections.get(window))
     ):
         return False
 
@@ -258,7 +243,7 @@ def _assign_syntax_for_new_view(view: sublime.View, event: ListenerEvent | None
 
 def _assign_syntax_for_st_syntax_test(view: sublime.View, event: ListenerEvent | None = None) -> bool:
     if (
-        (view_snapshot := ViewSnapshotCollection.get_by_view(view))
+        (view_snapshot := G.view_snapshot_collection.get_by_view(view))
         and (not view_snapshot.syntax or is_plaintext_syntax(view_snapshot.syntax))
         and (m := RE_ST_SYNTAX_TEST_LINE.search(view_snapshot.first_line))
         and (new_syntax := m.group("syntax")).endswith(".sublime-syntax")
@@ -288,7 +273,7 @@ def _assign_syntax_with_plugin_rules(
 
 
 def _assign_syntax_with_first_line(view: sublime.View, event: ListenerEvent | None = None) -> bool:
-    if not (view_snapshot := ViewSnapshotCollection.get_by_view(view)):
+    if not (view_snapshot := G.view_snapshot_collection.get_by_view(view)):
         return False
 
     # Note that this only works for files under some circumstances.
@@ -359,7 +344,7 @@ def _assign_syntax_with_trimmed_filename(view: sublime.View, event: ListenerEven
 
 def _assign_syntax_with_special_cases(view: sublime.View, event: ListenerEvent | None = None) -> bool:
     if not (
-        (view_snapshot := ViewSnapshotCollection.get_by_view(view))
+        (view_snapshot := G.view_snapshot_collection.get_by_view(view))
         and view_snapshot.syntax
         and is_plaintext_syntax(view_snapshot.syntax)
     ):
@@ -392,7 +377,7 @@ def is_json(view: sublime.View) -> bool:
 def _assign_syntax_with_guesslang_async(view: sublime.View, event: ListenerEvent | None = None) -> None:
     if not (
         G.guesslang_client
-        and (view_snapshot := ViewSnapshotCollection.get_by_view(view))
+        and (view_snapshot := G.view_snapshot_collection.get_by_view(view))
         # don't apply on those have an extension
         and (event == ListenerEvent.COMMAND or "." not in view_snapshot.file_name_unhidden)
         # only apply on plain text syntax

plugin/commands/auto_set_syntax_create_new_implementation.py

@@ -62,12 +62,10 @@ def _clone_file_as_template(
 
     new = window.new_file()
     new.run_command("append", {"characters": template})
-    new.settings().update(
-        {
-            "default_dir": save_dir,
-            "is_auto_set_syntax_template_buffer": True,
-        }
-    )
+    new.settings().update({
+        "default_dir": save_dir,
+        "is_auto_set_syntax_template_buffer": True,
+    })
 
     if syntax and (syntax := find_syntax_by_syntax_like(syntax)):
         new.assign_syntax(syntax)

plugin/commands/auto_set_syntax_debug_information.py

@@ -63,8 +63,8 @@ def run(self) -> None:
             },
         }
         info["plugin_settings"] = get_merged_plugin_settings(window=self.window)
-        info["syntax_rule_collection"] = G.get_syntax_rule_collection(self.window)
-        info["dropped_rules"] = G.get_dropped_rules(self.window)
+        info["syntax_rule_collection"] = G.syntax_rule_collections.get(self.window)
+        info["dropped_rules"] = G.dropped_rules_collection.get(self.window, [])
 
         msg = TEMPLATE.format_map(_pythonize(info))
 

plugin/guesslang/client.py

@@ -77,11 +77,9 @@ def request_guess_snapshot(
         if self.is_connected():
             assert self._ws
             self._ws.send(
-                sublime.encode_value(
-                    {
-                        "id": view_snapshot.id,
-                        "content": view_snapshot.content,
-                        "event_name": event.value if event else None,
-                    }
-                )
+                sublime.encode_value({
+                    "id": view_snapshot.id,
+                    "content": view_snapshot.content,
+                    "event_name": event.value if event else None,
+                })
             )

plugin/guesslang/server.py

@@ -14,18 +14,21 @@
 
 class GuesslangServer:
     SERVER_DIR: Final[Path] = PLUGIN_STORAGE_DIR / "guesslang-server"
-    SERVER_FILE: Final[Path] = PLUGIN_STORAGE_DIR / "guesslang-server/websocket.js"
+    SERVER_FILE: Final[Path] = SERVER_DIR / "websocket.js"
 
     def __init__(self, host: str, port: int) -> None:
         self.host = host
         self.port = port
-        # background server process(es)
-        self._subprocesses: set[subprocess.Popen] = set()
+        self._proc: subprocess.Popen | None = None
+        """The server process."""
 
     def start(self) -> bool:
         """Starts the guesslang server and return whether it starts."""
+        if self._proc:
+            Logger.log("⚠️ Server is already running.")
+            return True
         if not (node_info := parse_node_path_args()):
-            Logger.log("❌ Node.js binary is not found or not executable")
+            Logger.log("❌ Node.js binary is not found or not executable.")
             return False
         node_path, node_args = node_info
         Logger.log(f"✔ Use Node.js binary ({node_path}) and args ({node_args})")
@@ -42,35 +45,35 @@ def start(self) -> bool:
                 },
             )
         except Exception as e:
-            Logger.log(f"❌ Failed starting guesslang server because {e}")
+            Logger.log(f"❌ Failed starting guesslang server: {e}")
             return False
 
         if process.stdout and process.stdout.read(2) == "OK":
-            self._subprocesses.add(process)
+            self._proc = process
             return True
 
         Logger.log("❌ Failed starting guesslang server.")
         return False
 
     def stop(self) -> None:
-        for p in self._subprocesses:
-            try:
-                p.kill()
-            except Exception:
-                pass
-        for p in self._subprocesses:
-            try:
-                p.wait()
-            except Exception:
-                pass
-        self._subprocesses.clear()
+        if not self._proc:
+            return
+        try:
+            self._proc.kill()
+        except Exception:
+            pass
+        try:
+            self._proc.wait()
+        except Exception:
+            pass
+        self._proc = None
 
     def restart(self) -> bool:
         self.stop()
         return self.start()
 
     def is_running(self) -> bool:
-        return len(self._subprocesses) > 0
+        return self._proc is not None
 
     @staticmethod
     def _start_process(
@@ -85,6 +88,9 @@ def _start_process(
         else:
             startupinfo = None  # type: ignore
 
+        if isinstance(cmd, (str, Path)):
+            kwargs["shell"] = True
+
         return subprocess.Popen(
             cmd,
             startupinfo=startupinfo,
@@ -105,16 +111,16 @@ def parse_node_path_args() -> tuple[str, list[str]] | None:
             get_merged_plugin_setting("guesslang.node_bin_args"),
         ),
         ("${lsp_utils_node_bin}", []),
-        (shutil.which("electron"), []),
-        (shutil.which("node"), []),
-        (shutil.which("code"), ["--ms-enable-electron-run-as-node"]),  # VSCode
-        (shutil.which("codium"), []),  # VSCodium (non-Windows)
-        (shutil.which("VSCodium"), []),  # VSCodium (Windows)
+        ("electron", []),
+        ("node", []),
+        ("code", ["--ms-enable-electron-run-as-node"]),  # VSCode
+        ("codium", []),  # VSCodium (non-Windows)
+        ("VSCodium", []),  # VSCodium (Windows)
     ):
-        if (node := expand_variables(node)) and is_executable(node):
+        if (node := shutil.which(expand_variables(node)) or "") and is_executable(node):
             return (node, args)
     return None
 
 
 def is_executable(path: str | Path) -> bool:
-    return bool((os.path.isfile(path) and os.access(path, os.X_OK)))
+    return os.path.isfile(path) and os.access(path, os.X_OK)

plugin/listener.py

@@ -39,8 +39,8 @@ def set_up_window(window: sublime.Window) -> None:
 
 
 def tear_down_window(window: sublime.Window) -> None:
-    G.clear_syntax_rule_collection(window)
-    G.clear_dropped_rules(window)
+    G.syntax_rule_collections.pop(window, None)
+    G.dropped_rules_collection.pop(window, None)
     Logger.log("👋 Bye!", window=window)
     Logger.destroy(window=window)
 
@@ -59,11 +59,11 @@ def names_as_str(items: Iterable[Any], *, sep: str = ", ") -> str:
     Logger.log(f'🔍 Found "Constraint" implementations: {names_as_str(get_constraints())}', window=window)
 
     syntax_rule_collection = SyntaxRuleCollection.make(pref_syntax_rules(window=window))
-    G.set_syntax_rule_collection(window, syntax_rule_collection)
+    G.syntax_rule_collections[window] = syntax_rule_collection
     Logger.log(f"📜 Compiled syntax rule collection: {stringify(syntax_rule_collection)}", window=window)
 
-    dropped_rules = tuple(syntax_rule_collection.optimize())
-    G.set_dropped_rules(window, dropped_rules)
+    dropped_rules = list(syntax_rule_collection.optimize())
+    G.dropped_rules_collection[window] = dropped_rules
     Logger.log(f"✨ Optimized syntax rule collection: {stringify(syntax_rule_collection)}", window=window)
     Logger.log(f"💀 Dropped rules during optimizing: {stringify(dropped_rules)}", window=window)
 

plugin/logger.py

@@ -34,18 +34,16 @@ def _create_log_panel(window: sublime.Window) -> sublime.View:
     panel.assign_syntax(Logger.SYNTAX_FILE)
     panel.set_read_only(True)
     panel.set_scratch(True)
-    panel.settings().update(
-        {
-            "draw_white_space": "none",
-            "gutter": False,
-            "is_widget": True,  # ST 3 convention for a non-normal view
-            "line_numbers": False,
-            "scroll_context_lines": 0,
-            "scroll_past_end": False,
-            "spell_check": False,
-            "word_wrap": False,
-        }
-    )
+    panel.settings().update({
+        "draw_white_space": "none",
+        "gutter": False,
+        "is_widget": True,  # ST 3 convention for a non-normal view
+        "line_numbers": False,
+        "scroll_context_lines": 0,
+        "scroll_past_end": False,
+        "spell_check": False,
+        "word_wrap": False,
+    })
     return panel
 
 

plugin/rules/constraint.py

@@ -10,7 +10,8 @@
 
 from ..cache import clearable_lru_cache
 from ..constants import PLUGIN_NAME, ST_PLATFORM
-from ..snapshot import ViewSnapshot, ViewSnapshotCollection
+from ..shared import G
+from ..snapshot import ViewSnapshot
 from ..types import Optimizable, ST_ConstraintRule
 from ..utils import (
     camel_to_snake,
@@ -164,7 +165,7 @@ def _handled_case_insensitive(kwargs: dict[str, Any]) -> bool:
     @staticmethod
     def get_view_snapshot(view: sublime.View) -> ViewSnapshot:
         """Gets the cached information for the `view`."""
-        snapshot = ViewSnapshotCollection.get_by_view(view)
+        snapshot = G.view_snapshot_collection.get_by_view(view)
         assert snapshot  # our workflow guarantees this won't be None
         return snapshot
 
@@ -176,8 +177,10 @@ def find_parent_with_sibling(base: str | Path, sibling: str, *, use_exists: bool
 
         if use_exists:
             checker = Path.exists
+        elif sibling.endswith(("\\", "/")):
+            checker = Path.is_dir  # type: ignore
         else:
-            checker = Path.is_dir if sibling.endswith(("\\", "/")) else Path.is_file
+            checker = Path.is_file  # type: ignore
 
         return first_true(path.parents, pred=lambda p: checker(p / sibling))
 

plugin/rules/constraints/contains.py

@@ -13,7 +13,7 @@ class ContainsConstraint(AbstractConstraint):
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
 
-        self.needles: Tuple[str, ...] = self._handled_args()
+        self.needles: tuple[str, ...] = self._handled_args()
         self.threshold: int = kwargs.get("threshold", 1)
 
     def is_droppable(self) -> bool:

plugin/rules/constraints/first_line_contains.py

@@ -12,7 +12,7 @@ class FirstLineContainsConstraint(AbstractConstraint):
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
 
-        self.needles: Tuple[str, ...] = self._handled_args()
+        self.needles: tuple[str, ...] = self._handled_args()
 
     def is_droppable(self) -> bool:
         return not self.needles

plugin/rules/constraints/is_extension.py

@@ -20,7 +20,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
 
         self.case_insensitive = self._handled_case_insensitive(kwargs)
-        self.exts: Tuple[str, ...] = self._handled_args(_extensionize)
+        self.exts: tuple[str, ...] = self._handled_args(_extensionize)
         self.exts = tuple(map(self.fix_case, self.exts))
 
     def is_droppable(self) -> bool:

plugin/rules/constraints/is_in_git_repo.py

@@ -12,7 +12,7 @@
 class IsInGitRepoConstraint(AbstractConstraint):
     """Check whether this file is in a git repo."""
 
-    _success_dirs: Set[Path] = set()
+    _success_dirs: set[Path] = set()
     """Cached directories which make the result `True`."""
 
     def test(self, view: sublime.View) -> bool:

plugin/rules/constraints/is_in_hg_repo.py

@@ -12,7 +12,7 @@
 class IsInHgRepoConstraint(AbstractConstraint):
     """Check whether this file is in a Mercurial repo."""
 
-    _success_dirs: Set[Path] = set()
+    _success_dirs: set[Path] = set()
     """Cached directories which make the result `True`."""
 
     def test(self, view: sublime.View) -> bool: