port remaining CSVs from resource.open to resource.watch (talonhub#1555)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Nicholas Riley <[email protected]>

BREAKING_CHANGES.txt

@@ -7,6 +7,9 @@ and when the change was applied given the delay between changes being
 submitted and the time they were reviewed and merged.
 
 ---
+* 2024-09-07 Removed `get_list_from_csv` from `user_settings.py`. Please
+  use the new `track_csv_list` decorator, which leverages Talon's
+  `talon.watch` API for robustness on Talon launch.
 * 2024-09-07 If you've updated `community` since 2024-08-31, you may
   need to replace `host:` with `hostname:` in the header of
   `core/system_paths-<hostname>.talon-list` due to an issue with

apps/emacs/emacs_commands.py

@@ -33,10 +33,9 @@ def emacs_command_short_form(command_name: str) -> Optional[str]:
         return emacs_commands.get(command_name, Command(command_name)).short
 
 
-def load_csv():
-    filepath = Path(__file__).parents[0] / "emacs_commands.csv"
-    with resource.open(filepath) as f:
-        rows = list(csv.reader(f))
+@resource.watch("emacs_commands.csv")
+def load_commands(f):
+    rows = list(csv.reader(f))
     # Check headers
     assert rows[0] == ["Command", " Key binding", " Short form", " Spoken form"]
 
@@ -46,7 +45,7 @@ def load_csv():
             continue
         if len(row) > 4:
             print(
-                f'"{filepath}": More than four values in row: {row}. '
+                f"emacs_commands.csv: More than four values in row: {row}. "
                 + " Ignoring the extras"
             )
         name, keys, short, spoken = (
@@ -70,7 +69,3 @@ def load_csv():
             if c.spoken:
                 command_list[c.spoken] = c.name
     ctx.lists["self.emacs_command"] = command_list
-
-
-# TODO: register on change to file!
-app.register("ready", load_csv)

core/abbreviate/abbreviate.py

@@ -2,12 +2,13 @@
 
 from talon import Context, Module
 
-from ..user_settings import get_list_from_csv
+from ..user_settings import track_csv_list
 
 mod = Module()
+ctx = Context()
 mod.list("abbreviation", desc="Common abbreviation")
 
-
+abbreviations_list = {}
 abbreviations = {
     "J peg": "jpg",
     "abbreviate": "abbr",
@@ -447,24 +448,26 @@
     "work in progress": "wip",
 }
 
-# This variable is also considered exported for the create_spoken_forms module
-abbreviations_list = get_list_from_csv(
-    "abbreviations.csv",
-    headers=("Abbreviation", "Spoken Form"),
-    default=abbreviations,
+
+@track_csv_list(
+    "abbreviations.csv", headers=("Abbreviation", "Spoken Form"), default=abbreviations
 )
+def on_abbreviations(values):
+    global abbreviations_list
 
-# Matches letters and spaces, as currently, Talon doesn't accept other characters in spoken forms.
-PATTERN = re.compile(r"^[a-zA-Z ]+$")
-abbreviation_values = {
-    v: v for v in abbreviations_list.values() if PATTERN.match(v) is not None
-}
+    # note: abbreviations_list is  imported by the create_spoken_forms module
+    abbreviations_list = values
 
-# Allows the abbreviated/short form to be used as spoken phrase. eg "brief app" -> app
-abbreviations_list_with_values = {
-    **abbreviation_values,
-    **abbreviations_list,
-}
+    # Matches letters and spaces, as currently, Talon doesn't accept other characters in spoken forms.
+    PATTERN = re.compile(r"^[a-zA-Z ]+$")
+    abbreviation_values = {
+        v: v for v in abbreviations_list.values() if PATTERN.match(v) is not None
+    }
 
-ctx = Context()
-ctx.lists["user.abbreviation"] = abbreviations_list_with_values
+    # Allows the abbreviated/short form to be used as spoken phrase. eg "brief app" -> app
+    abbreviations_list_with_values = {
+        **{v: v for v in abbreviation_values.values()},
+        **abbreviations_list,
+    }
+
+    ctx.lists["user.abbreviation"] = abbreviations_list_with_values

core/create_spoken_forms.py

@@ -6,34 +6,58 @@
 
 from talon import Module, actions
 
-from .abbreviate.abbreviate import abbreviations_list
-from .file_extension.file_extension import file_extensions
 from .keys.keys import symbol_key_words
 from .numbers.numbers import digits_map, scales, teens, tens
+from .user_settings import track_csv_list
 
 mod = Module()
 
-
 DEFAULT_MINIMUM_TERM_LENGTH = 2
 EXPLODE_MAX_LEN = 3
 FANCY_REGULAR_EXPRESSION = r"[A-Z]?[a-z]+|[A-Z]+(?![a-z])|[0-9]+"
-FILE_EXTENSIONS_REGEX = "|".join(
-    re.escape(file_extension.strip()) + "$"
-    for file_extension in file_extensions.values()
-)
 SYMBOLS_REGEX = "|".join(re.escape(symbol) for symbol in set(symbol_key_words.values()))
-REGEX_NO_SYMBOLS = re.compile(
-    "|".join(
-        [
-            FANCY_REGULAR_EXPRESSION,
-            FILE_EXTENSIONS_REGEX,
-        ]
+FILE_EXTENSIONS_REGEX = r"^\b$"
+file_extensions = {}
+
+
+def update_regex():
+    global REGEX_NO_SYMBOLS
+    global REGEX_WITH_SYMBOLS
+    REGEX_NO_SYMBOLS = re.compile(
+        "|".join(
+            [
+                FANCY_REGULAR_EXPRESSION,
+                FILE_EXTENSIONS_REGEX,
+            ]
+        )
+    )
+    REGEX_WITH_SYMBOLS = re.compile(
+        "|".join([FANCY_REGULAR_EXPRESSION, FILE_EXTENSIONS_REGEX, SYMBOLS_REGEX])
+    )
+
+
+update_regex()
+
+
+@track_csv_list("file_extensions.csv", headers=("File extension", "Name"))
+def on_extensions(values):
+    global FILE_EXTENSIONS_REGEX
+    global file_extensions
+    file_extensions = values
+    FILE_EXTENSIONS_REGEX = "|".join(
+        re.escape(file_extension.strip()) + "$" for file_extension in values.values()
     )
-)
+    update_regex()
+
+
+abbreviations_list = {}
+
+
+@track_csv_list("abbreviations.csv", headers=("Abbreviation", "Spoken Form"))
+def on_abbreviations(values):
+    global abbreviations_list
+    abbreviations_list = values
 
-REGEX_WITH_SYMBOLS = re.compile(
-    "|".join([FANCY_REGULAR_EXPRESSION, FILE_EXTENSIONS_REGEX, SYMBOLS_REGEX])
-)
 
 REVERSE_PRONUNCIATION_MAP = {
     **{str(value): key for key, value in digits_map.items()},

core/file_extension/file_extension.py

@@ -1,6 +1,6 @@
 from talon import Context, Module
 
-from ..user_settings import get_list_from_csv
+from ..user_settings import track_csv_list
 
 mod = Module()
 mod.list("file_extension", desc="A file extension, such as .py")
@@ -55,11 +55,13 @@
     "dot log": ".log",
 }
 
-file_extensions = get_list_from_csv(
+ctx = Context()
+
+
+@track_csv_list(
     "file_extensions.csv",
     headers=("File extension", "Name"),
     default=_file_extensions_defaults,
 )
-
-ctx = Context()
-ctx.lists["self.file_extension"] = file_extensions
+def on_update(values):
+    ctx.lists["self.file_extension"] = values

core/keys/keys.py

@@ -1,7 +1,5 @@
 from talon import Context, Module, app
 
-from ..user_settings import get_list_from_csv
-
 # used for number keys & function keys respectively
 digits = "zero one two three four five six seven eight nine".split()
 f_digits = "one two three four five six seven eight nine ten eleven twelve thirteen fourteen fifteen sixteen seventeen eighteen nineteen twenty".split()

core/user_settings.py

@@ -1,43 +1,31 @@
 import csv
 import os
 from pathlib import Path
+from typing import IO, Callable
 
 from talon import resource
 
 # NOTE: This method requires this module to be one folder below the top-level
 #   community/knausj folder.
 SETTINGS_DIR = Path(__file__).parents[1] / "settings"
+SETTINGS_DIR.mkdir(exist_ok=True)
 
-if not SETTINGS_DIR.is_dir():
-    os.mkdir(SETTINGS_DIR)
+CallbackT = Callable[[dict[str, str]], None]
+DecoratorT = Callable[[CallbackT], CallbackT]
 
 
-def get_list_from_csv(
-    filename: str, headers: tuple[str, str], default: dict[str, str] = {}
-):
-    """Retrieves list from CSV"""
-    path = SETTINGS_DIR / filename
-    assert filename.endswith(".csv")
-
-    if not path.is_file():
-        with open(path, "w", encoding="utf-8", newline="") as file:
-            writer = csv.writer(file)
-            writer.writerow(headers)
-            for key, value in default.items():
-                writer.writerow([key] if key == value else [value, key])
-
-    # Now read via resource to take advantage of talon's
-    # ability to reload this script for us when the resource changes
-    with resource.open(str(path), "r") as f:
-        rows = list(csv.reader(f))
+def read_csv_list(
+    f: IO, headers: tuple[str, str], is_spoken_form_first: bool = False
+) -> dict[str, str]:
+    rows = list(csv.reader(f))
 
     # print(str(rows))
     mapping = {}
     if len(rows) >= 2:
         actual_headers = rows[0]
         if not actual_headers == list(headers):
             print(
-                f'"{filename}": Malformed headers - {actual_headers}.'
+                f'"{f.name}": Malformed headers - {actual_headers}.'
                 + f" Should be {list(headers)}. Ignoring row."
             )
         for row in rows[1:]:
@@ -47,10 +35,14 @@ def get_list_from_csv(
             if len(row) == 1:
                 output = spoken_form = row[0]
             else:
-                output, spoken_form = row[:2]
+                if is_spoken_form_first:
+                    spoken_form, output = row[:2]
+                else:
+                    output, spoken_form = row[:2]
+
                 if len(row) > 2:
                     print(
-                        f'"{filename}": More than two values in row: {row}.'
+                        f'"{f.name}": More than two values in row: {row}.'
                         + " Ignoring the extras."
                     )
             # Leading/trailing whitespace in spoken form can prevent recognition.
@@ -60,6 +52,44 @@ def get_list_from_csv(
     return mapping
 
 
+def write_csv_defaults(
+    path: Path,
+    headers: tuple[str, str],
+    default: dict[str, str] = None,
+    is_spoken_form_first: bool = False,
+) -> None:
+    if not path.is_file() and default is not None:
+        with open(path, "w", encoding="utf-8") as file:
+            writer = csv.writer(file)
+            writer.writerow(headers)
+            for key, value in default.items():
+                if key == value:
+                    writer.writerow([key])
+                elif is_spoken_form_first:
+                    writer.writerow([key, value])
+                else:
+                    writer.writerow([value, key])
+
+
+def track_csv_list(
+    filename: str,
+    headers: tuple[str, str],
+    default: dict[str, str] = None,
+    is_spoken_form_first: bool = False,
+) -> DecoratorT:
+    assert filename.endswith(".csv")
+    path = SETTINGS_DIR / filename
+    write_csv_defaults(path, headers, default, is_spoken_form_first)
+
+    def decorator(fn: CallbackT) -> CallbackT:
+        @resource.watch(str(path))
+        def on_update(f):
+            data = read_csv_list(f, headers, is_spoken_form_first)
+            fn(data)
+
+    return decorator
+
+
 def append_to_csv(filename: str, rows: dict[str, str]):
     path = SETTINGS_DIR / filename
     assert filename.endswith(".csv")