port from resource.open to resource.watch

apps/emacs/emacs_commands.py

@@ -33,11 +33,10 @@ def __init__(self):
 emacs_commands = CommandInfo()
 
 
-def load_csv():
+@resource.watch("emacs_commands.csv")
+def load_commands(f):
     global emacs_commands
-    filepath = Path(__file__).parents[0] / "emacs_commands.csv"
-    with resource.open(filepath) as f:
-        rows = list(csv.reader(f))
+    rows = list(csv.reader(f))
     # Check headers
     assert rows[0] == ["Command", " Key binding", " Short form", " Spoken form"]
 
@@ -76,10 +75,6 @@ def load_csv():
     ctx.lists["self.emacs_command"] = info.by_spoken
 
 
-# TODO: register on change to file!
-app.register("ready", load_csv)
-
-
 @mod.action_class
 class Actions:
     def emacs(command_name: str, prefix: Optional[int] = None):

apps/git/git.py

@@ -1,6 +1,7 @@
+from pathlib import Path
+from typing import IO
 import csv
 import os
-from pathlib import Path
 
 from talon import Context, Module, actions, resource
 
@@ -10,14 +11,8 @@
 mod.list("git_command", desc="Git commands.")
 mod.list("git_argument", desc="Command-line git options and arguments.")
 
-dirpath = Path(__file__).parent
-arguments_csv_path = str(dirpath / "git_arguments.csv")
-commands_csv_path = str(dirpath / "git_commands.csv")
-
-
-def make_list(path):
-    with resource.open(path, "r") as f:
-        rows = list(csv.reader(f))
+def make_list(f: IO) -> dict[str, str]:
+    rows = list(csv.reader(f))
     mapping = {}
     # ignore header row
     for row in rows[1:]:
@@ -32,10 +27,13 @@ def make_list(path):
         mapping[spoken_form] = output
     return mapping
 
+@resource.watch("git_arguments.csv")
+def load_arguments(f):
+    ctx.lists["self.git_argument"] = make_list(f)
 
-ctx.lists["self.git_argument"] = make_list(arguments_csv_path)
-ctx.lists["self.git_command"] = make_list(commands_csv_path)
-
+@resource.watch("git_commands.csv")
+def load_commands(f):
+    ctx.lists["self.git_command"] = make_list(f)
 
 @mod.capture(rule="{user.git_argument}+")
 def git_arguments(m) -> str:

core/abbreviate/abbreviate.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("abbreviation", desc="Common abbreviation")
@@ -446,17 +446,18 @@
 }
 
 # 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,
-)
-
-# Allows the abbreviated/short form to be used as spoken phrase. eg "brief app" -> app
-abbreviations_list_with_values = {
-    **{v: v for v in abbreviations_list.values()},
-    **abbreviations_list,
-}
+abbreviations_list = {}
 
 ctx = Context()
-ctx.lists["user.abbreviation"] = abbreviations_list_with_values
+
+@track_csv_list("abbreviations.csv", headers=("Abbreviation", "Spoken Form"), default=abbreviations)
+def on_abbreviations(values):
+    global abbreviations_list
+    abbreviations_list = 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 abbreviations_list.values()},
+        **abbreviations_list,
+    }
+    ctx.lists["user.abbreviation"] = abbreviations_list_with_values

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")
@@ -52,11 +52,8 @@
     "dot toml": ".toml",
 }
 
-file_extensions = get_list_from_csv(
-    "file_extensions.csv",
-    headers=("File extension", "Name"),
-    default=_file_extensions_defaults,
-)
-
 ctx = Context()
-ctx.lists["self.file_extension"] = file_extensions
+
+@track_csv_list("file_extensions.csv", headers=("File extension", "Name"), default=_file_extensions_defaults)
+def on_update(values):
+    ctx.lists["self.file_extension"] = file_extensions

core/keys/keys.py

@@ -1,6 +1,6 @@
 from talon import Context, Module, actions, app
 
-from ..user_settings import get_list_from_csv
+from ..user_settings import track_csv_list
 
 
 def setup_default_alphabet():
@@ -16,10 +16,6 @@ def setup_default_alphabet():
     return initial_default_alphabet_dict
 
 
-alphabet_list = get_list_from_csv(
-    "alphabet.csv", ("Letter", "Spoken Form"), setup_default_alphabet()
-)
-
 # 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".split()
@@ -132,7 +128,10 @@ def letters(m) -> str:
     modifier_keys["command"] = "cmd"
     modifier_keys["option"] = "alt"
 ctx.lists["self.modifier_key"] = modifier_keys
-ctx.lists["self.letter"] = alphabet_list
+
+@track_csv_list("alphabet.csv", ("Letter", "Spoken Form"), setup_default_alphabet())
+def on_alphabet(values):
+    ctx.lists["self.letter"] = values
 
 # `punctuation_words` is for words you want available BOTH in dictation and as key names in command mode.
 # `symbol_key_words` is for key names that should be available in command mode, but NOT during dictation.

core/system_paths.py

@@ -7,7 +7,7 @@
 
 from talon import Context, Module, actions, app
 
-from .user_settings import get_list_from_csv
+from .user_settings import track_csv_list
 
 mod = Module()
 ctx = Context()
@@ -49,11 +49,9 @@ def on_ready():
 
             default_system_paths.update(onedrive_paths)
 
-    system_paths = get_list_from_csv(
-        "system_paths.csv", headers=("Path", "Spoken"), default=default_system_paths
-    )
-
-    ctx.lists["user.system_paths"] = system_paths
+    @track_csv_list("system_paths.csv", headers=("Path", "Spoken"), default=default_system_paths)
+    def on_csv(system_paths):
+        ctx.lists["user.system_paths"] = system_paths
 
 
 @mod.capture(rule="{user.system_paths}")

core/user_settings.py

@@ -1,6 +1,7 @@
+from pathlib import Path
+from typing import Callable
 import csv
 import os
-from pathlib import Path
 
 from talon import resource
 
@@ -12,24 +13,11 @@
     os.mkdir(SETTINGS_DIR)
 
 
-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")
+CallbackT  = Callable[[dict[str, str]], None]
+DecoratorT = Callable[[DecoratedT], DecoratedT]
 
-    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]) -> dict[str, str]:
+    rows = list(csv.reader(f))
 
     # print(str(rows))
     mapping = {}
@@ -59,6 +47,40 @@ def get_list_from_csv(
 
     return mapping
 
+def write_csv_defaults(path: Path, headers: tuple[str, str], default: dict[str, str]=None) -> None:
+    if not path.is_file() and default is not None:
+        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])
+
+def track_csv_list(filename: str, headers: tuple[str, str], default: dict[str, str]=None) -> DecoratorT:
+    assert filename.endswith(".csv")
+    path = SETTINGS_DIR / filename
+    write_csv_defaults(path, headers, default)
+
+    def decorator(fn: CallbackT) -> CallbackT:
+        @resource.watch(str(path))
+        def on_update(f):
+            data = read_csv_list(f, headers, default)
+            fn(data)
+
+    return decorator
+
+# NOTE: this is deprecated, use @track_csv_list instead.
+def get_list_from_csv(
+    filename: str, headers: tuple[str, str], default: dict[str, str] = {}
+):
+    """Retrieves list from CSV"""
+    assert filename.endswith(".csv")
+    path = SETTINGS_DIR / filename
+    write_csv_defaults(path, headers, default)
+
+    # 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:
+        return read_csv_list(f, headers)
 
 def append_to_csv(filename: str, rows: dict[str, str]):
     path = SETTINGS_DIR / filename

core/vocabulary/vocabulary.py

@@ -5,7 +5,7 @@
 from talon import Context, Module, actions
 from talon.grammar import Phrase
 
-from ..user_settings import append_to_csv, get_list_from_csv
+from ..user_settings import append_to_csv, track_csv_list
 
 mod = Module()
 ctx = Context()
@@ -49,39 +49,20 @@
 # implementation of `dictate.replace_words` (at bottom of file) to rewrite words
 # and phrases Talon recognized. This does not change the priority with which
 # Talon recognizes particular phrases over others.
-phrases_to_replace = get_list_from_csv(
-    "words_to_replace.csv",
-    headers=("Replacement", "Original"),
-    default=_word_map_defaults,
-)
-
-# "dictate.word_map" is used by Talon's built-in default implementation of
-# `dictate.replace_words`, but supports only single-word replacements.
-# Multi-word phrases are ignored.
-ctx.settings["dictate.word_map"] = phrases_to_replace
-
+@track_csv_list("words_to_replace.csv", headers=("Replacement", "Original"), default=_word_map_defaults)
+def on_word_map(values):
+    # "dictate.word_map" is used by Talon's built-in default implementation of
+    # `dictate.replace_words`, but supports only single-word replacements.
+    # Multi-word phrases are ignored.
+    ctx.settings["dictate.word_map"] = values
 
-# Default words that should be added to Talon's vocabulary.
-# Don't edit this. Edit 'additional_vocabulary.csv' instead
-_simple_vocab_default = ["nmap", "admin", "Cisco", "Citrix", "VPN", "DNS", "Minecraft"]
-
-# Defaults for different pronounciations of words that need to be added to
-# Talon's vocabulary.
-_default_vocabulary = {
-    "N map": "nmap",
-    "under documented": "under-documented",
-}
-_default_vocabulary.update({word: word for word in _simple_vocab_default})
 
 # "user.vocabulary" is used to explicitly add words/phrases that Talon doesn't
 # recognize. Words in user.vocabulary (or other lists and captures) are
 # "command-like" and their recognition is prioritized over ordinary words.
-vocabulary = get_list_from_csv(
-    "additional_words.csv",
-    headers=("Word(s)", "Spoken Form (If Different)"),
-    default=_default_vocabulary,
-)
-ctx.lists["user.vocabulary"] = vocabulary
+@track_csv_list("additional_words.csv", headers=("Word(s)", "Spoken Form (If Different)"))
+def on_vocab(values):
+    ctx.lists["user.vocabulary"] = values
 
 
 class PhraseReplacer:

core/websites_and_search_engines/websites_and_search_engines.py

@@ -3,7 +3,7 @@
 
 from talon import Context, Module
 
-from ..user_settings import get_list_from_csv
+from ..user_settings import track_csv_list
 
 mod = Module()
 mod.list("website", desc="A website.")
@@ -40,17 +40,14 @@
 }
 
 ctx = Context()
-ctx.lists["self.website"] = get_list_from_csv(
-    "websites.csv",
-    headers=("URL", "Spoken name"),
-    default=website_defaults,
-)
-ctx.lists["self.search_engine"] = get_list_from_csv(
-    "search_engines.csv",
-    headers=("URL Template", "Name"),
-    default=_search_engine_defaults,
-)
 
+@track_csv_list("websites.csv", headers=("URL", "Spoken name"), default=website_defaults)
+def on_websites(values):
+    ctx.lists["self.website"] = values
+
+@track_csv_list("search_engines.csv", headers=("URL Template", "Name"), default=_search_engine_defaults)
+def on_search_engines(values):
+    ctx.lists["self.search_engines"] = values
 
 @mod.action_class
 class Actions:

tags/terminal/unix_utilities.py

@@ -1,6 +1,6 @@
 from talon import Context, Module
 
-from ...core.user_settings import get_list_from_csv
+from ...core.user_settings import track_csv_list
 
 ctx = Context()
 mod = Module()
@@ -75,11 +75,8 @@
     "who am I": "whoami",
 }
 
-unix_utilities = get_list_from_csv(
-    "unix_utilities.csv",
-    headers=("command", "spoken"),
-    default=default_unix_utilities,
-)
-
 mod.list("unix_utility", desc="A common utility command")
-ctx.lists["self.unix_utility"] = unix_utilities
+
+@track_csv_list("unix_utilities.csv", headers=("command", "spoken"), default=default_unix_utilities)
+def on_utilities(values):
+    ctx.lists["self.unix_utility"] = values