Add command block to execute shell commands (#124)

Author: mandel

pdl-live/src/pdl_ast.d.ts

@@ -2163,7 +2163,7 @@ export type Kind15 = "code";
  * Programming language of the code.
  *
  */
-export type Lang = "python";
+export type Lang = "python" | "command";
 /**
  * Code to execute.
  *

src/pdl/pdl-schema.json

@@ -2555,10 +2555,10 @@
           "type": "string"
         },
         "lang": {
-          "const": "python",
           "description": "Programming language of the code.\n    ",
           "enum": [
-            "python"
+            "python",
+            "command"
           ],
           "title": "Lang",
           "type": "string"

src/pdl/pdl_ast.py

@@ -278,7 +278,7 @@ class CodeBlock(Block):
     """Execute a piece of code."""
 
     kind: Literal[BlockKind.CODE] = BlockKind.CODE
-    lang: Literal["python"]
+    lang: Literal["python", "command"]
     """Programming language of the code.
     """
     code: "BlocksType"

src/pdl/pdl_interpreter.py

@@ -1,6 +1,8 @@
 import json
 import logging
 import re
+import shlex
+import subprocess
 import sys
 import types
 
@@ -1222,6 +1224,16 @@ def step_call_code(
                     loc=loc,
                     trace=block.model_copy(update={"code": code_s}),
                 ) from exc
+        case "command":
+            try:
+                result = call_command(code_s)
+                background = [{"role": state.role, "content": result}]
+            except Exception as exc:
+                raise PDLRuntimeError(
+                    f"Code error: {repr(exc)}",
+                    loc=loc,
+                    trace=block.model_copy(update={"code": code_s}),
+                ) from exc
         case _:
             message = f"Unsupported language: {block.lan}"
             raise PDLRuntimeError(
@@ -1244,6 +1256,17 @@ def call_python(code: str, scope: dict) -> Any:
     return result
 
 
+def call_command(code: str) -> str:
+    args = shlex.split(code)
+    p = subprocess.run(args, capture_output=True, text=True, check=False)
+    if p.stderr != "":
+        print(p.stderr, file=sys.stderr)
+    if p.returncode != 0:
+        raise ValueError(f"command exited with non zero code: {p.returncode}")
+    output = p.stdout
+    return output
+
+
 def step_call(
     state: InterpreterState, scope: ScopeType, block: CallBlock, loc: LocationType
 ) -> Generator[YieldMessage, Any, tuple[Any, Messages, ScopeType, CallBlock]]:

tests/test_code.py

@@ -57,3 +57,17 @@ def test_contribute_false():
     data = Program.model_validate(show_result_data([]))
     text, _, _, _ = process_prog(state, empty_scope, data)
     assert text == ""
+
+
+command_data = [
+    {"def": "world", "lang": "command", "code": "echo -n World", "contribute": []},
+    "Hello ${ world }!",
+]
+
+
+def test_command():
+    state = InterpreterState()
+    data = Program.model_validate(command_data)
+    document, _, scope, _ = process_prog(state, empty_scope, data)
+    assert document == "Hello World!"
+    assert scope["world"] == "World"