initial Lexurgy client

Author: neta-elad

pyconlang/assets/__init__.py

@@ -1 +1 @@
-LEXURGY_VERSION = "1.2.1"
+LEXURGY_VERSION = "1.2.2-server"

pyconlang/assets/lexurgy-1.2.2-server.zip

@@ -5,17 +5,21 @@
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field
 from functools import cached_property
-from itertools import chain
 from pathlib import Path
-from subprocess import run
-from time import time
 from unicodedata import normalize
 
 from .. import PYCONLANG_PATH
 from ..assets import LEXURGY_VERSION
 from ..checksum import checksum
 from ..domain import Component, Morpheme, ResolvedForm
-from ..lexurgy.domain import TraceLine
+from ..lexurgy import client
+from ..lexurgy.client import LexurgyClient
+from ..lexurgy.domain import (
+    LexurgyErrorResponse,
+    LexurgyRequest,
+    LexurgyResponse,
+    TraceLine,
+)
 from ..lexurgy.tracer import parse_trace_lines
 from .arrange import AffixArranger
 from .batch import Batcher, Cache, ComponentQuery, CompoundQuery, Query
@@ -59,6 +63,10 @@ class Evolver:
     def arranger(self) -> AffixArranger:
         return AffixArranger.from_path(CHANGES_PATH)
 
+    @cached_property
+    def lexurgy(self) -> LexurgyClient:
+        return client()  # .value
+
     @classmethod
     def load(cls) -> "Evolver":
         if not all(
@@ -188,67 +196,41 @@ def evolve_words(
         if not words:
             return [], {}
 
-        base_name = f"words-{time():.0f}"
-
-        self.evolve_directory.mkdir(parents=True, exist_ok=True)
-
-        input_words = self.evolve_directory / f"{base_name}.wli"
-        input_words.write_text("\n".join(words))
-
-        output_words = self.evolve_directory / f"{base_name}_ev.wli"
-        output_words.unlink(missing_ok=True)
-
-        phonetic_words = self.evolve_directory / f"{base_name}_phonetic.wli"
-        phonetic_words.unlink(missing_ok=True)
-
-        trace_file = self.evolve_directory / f"{base_name}_trace.wli"
-        trace_file.unlink(missing_ok=True)
-
-        args = [
-            "sh",
-            str(LEXURGY_PATH),
-            "sc",
-            str(CHANGES_PATH),
-            str(input_words),
-            "-m",
-        ]
-
-        if start is not None:
-            args.append("-a")
-            args.append(start)
+        debug_words = []
+        if trace:
+            debug_words = words
 
-        if end is not None:
-            args.append("-b")
-            args.append(end)
+        request = LexurgyRequest(words, start, end, debug_words)
 
-        if trace:
-            args.extend(chain(*zip(["-t"] * len(words), words)))
+        response = self.lexurgy.roundtrip(request)
 
-        result = run(args, capture_output=True, text=True)
+        match response:
+            case LexurgyErrorResponse():
+                raise LexurgyError(response.message)
 
-        if result.returncode != 0:
-            # todo too heuristic?
-            stdout = result.stdout.strip().splitlines()
-            if len(stdout) > 0:
-                raise LexurgyError(result.stdout.strip().splitlines()[-1])
-            else:
-                raise LexurgyError(result.stderr)
+            case LexurgyResponse():
+                moderns = [normalize("NFD", word) for word in response.words]
 
-        moderns = normalize("NFD", output_words.read_text().strip()).split("\n")
+                if "phonetic" in response.intermediates:
+                    phonetics = [
+                        normalize("NFD", word)
+                        for word in response.intermediates["phonetic"]
+                    ]
+                else:
+                    phonetics = moderns
 
-        if phonetic_words.exists():
-            phonetics = normalize("NFD", phonetic_words.read_text()).strip().split("\n")
-        else:
-            phonetics = moderns
+                assert len(phonetics) == len(moderns)
 
-        trace_lines: Mapping[str, list[TraceLine]] = {}
-        if trace:
-            trace_lines = parse_trace_lines(trace_file.read_text(), words[0])
+                trace_lines: Mapping[str, list[TraceLine]] = {}
+                if trace:
+                    trace_lines = parse_trace_lines(
+                        "\n".join(response.trace_lines), words[0]
+                    )
 
-        return [
-            Evolved(proto, modern, phonetic)
-            for proto, modern, phonetic in zip(words, moderns, phonetics)
-        ], trace_lines
+                return [
+                    Evolved(proto, modern, phonetic)
+                    for proto, modern, phonetic in zip(words, moderns, phonetics)
+                ], trace_lines
 
     def cleanup(self) -> None:
         if self.evolve_directory.exists():

pyconlang/evolve/__init__.py

@@ -0,0 +1,19 @@
+from pathlib import Path
+from subprocess import PIPE, Popen
+
+from .. import PYCONLANG_PATH
+from ..assets import LEXURGY_VERSION
+from .client import LexurgyClient
+
+LEXURGY_PATH = PYCONLANG_PATH / f"lexurgy-{LEXURGY_VERSION}" / "bin" / "lexurgy"
+CHANGES_PATH = Path("changes.lsc")
+
+
+def client() -> LexurgyClient:
+    args = [
+        "sh",
+        str(LEXURGY_PATH),
+        "server",
+        str(CHANGES_PATH),
+    ]
+    return LexurgyClient(Popen(args, stdin=PIPE, stdout=PIPE, text=True, bufsize=1))

pyconlang/lexurgy/__init__.py

@@ -0,0 +1,46 @@
+from dataclasses import dataclass
+from functools import cached_property
+from subprocess import Popen
+from typing import IO
+
+from .domain import (
+    AnyLexurgyResponse,
+    LexurgyErrorResponse,
+    LexurgyRequest,
+    LexurgyResponse,
+)
+
+
+@dataclass
+class LexurgyClient:
+    popen: Popen[str]
+
+    @cached_property
+    def stdin(self) -> IO[str]:
+        assert self.popen.stdin is not None
+        return self.popen.stdin
+
+    @cached_property
+    def stdout(self) -> IO[str]:
+        assert self.popen.stdout is not None
+        return self.popen.stdout
+
+    def write_line(self, line: str) -> None:
+        self.stdin.write(f"{line}\n")
+
+    def read_line(self) -> str:
+        return self.stdout.readline()
+
+    def send(self, request: LexurgyRequest) -> None:
+        self.write_line(request.to_json())
+
+    def receive(self) -> AnyLexurgyResponse:
+        line = self.read_line()
+        try:
+            return LexurgyResponse.from_json(line)
+        except:
+            return LexurgyErrorResponse.from_json(line)
+
+    def roundtrip(self, request: LexurgyRequest) -> AnyLexurgyResponse:
+        self.send(request)
+        return self.receive()

pyconlang/lexurgy/client.py

@@ -1,4 +1,7 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from typing import cast
+
+from dataclasses_json import DataClassJsonMixin, LetterCase, config
 
 
 @dataclass(eq=True, frozen=True)
@@ -10,3 +13,35 @@ class TraceLine:
 
     def set_word(self, new_word: str) -> "TraceLine":
         return TraceLine(self.rule, new_word, self.before, self.after)
+
+
+class CamelCaseJsonMixin(DataClassJsonMixin):
+    dataclass_json_config = cast(
+        None, config(letter_case=LetterCase.CAMEL)["dataclasses_json"]
+    )
+
+
+# @dataclass_json(letter_case=LetterCase.CAMEL)
+@dataclass
+class LexurgyRequest(CamelCaseJsonMixin):
+    words: list[str]
+    start_at: str | None = field(default=None)
+    stop_before: str | None = field(default=None)
+    debug_words: list[str] = field(default_factory=list)
+    romanize: bool = field(default=True)
+
+
+@dataclass
+class LexurgyResponse(CamelCaseJsonMixin):
+    words: list[str]
+    intermediates: dict[str, list[str]]
+    trace_lines: list[str]
+
+
+@dataclass
+class LexurgyErrorResponse(CamelCaseJsonMixin):
+    message: str
+    stack_trace: list[str]
+
+
+AnyLexurgyResponse = LexurgyResponse | LexurgyErrorResponse

pyconlang/lexurgy/domain.py

@@ -1,5 +1,3 @@
-from typing import Optional
-
 from wcwidth import wcswidth
 
 from .metadata import Metadata
@@ -23,7 +21,7 @@ def center(string: str, width: int, fill_char: str = " ") -> str:
     return fill_char * add_left + string + fill_char * add_right
 
 
-def combine(head: str, tail: str, syllable_break: Optional[str] = None) -> str:
+def combine(head: str, tail: str, syllable_break: str | None = None) -> str:
     if syllable_break is None:
         syllable_break = default_syllable_break()
 

pyconlang/unicode.py

@@ -19,7 +19,8 @@ dependencies = [
     "toml==0.10.2",
     "pyparsing==3.0.9",
     "unidecode==1.3.4",
-    "wcwidth==0.2.5 "
+    "wcwidth==0.2.5",
+    "dataclasses-json==0.5.8"
 ]
 dynamic = ["version"]