beam: safe arithmetic expression evaluator with tests

wizard-work/beam/expr_eval.py

@@ -0,0 +1,164 @@
+"""Tiny arithmetic expression evaluator with proper precedence.
+
+Supports: + - * / // % **, unary +/-, parentheses, variables (dict),
+functions (sin, cos, sqrt, abs, max, min), and error reporting with
+position. No eval/exec — pure recursive descent.
+"""
+import math
+import re
+
+class EvalError(Exception):
+    def __init__(self, msg, pos=None):
+        super().__init__(msg)
+        self.pos = pos
+
+_TOKEN_RE = re.compile(r"""
+    \s*(?:
+        (?P<num>\d+(?:\.\d+)?(?:[eE][+-]?\d+)?)
+      | (?P<ident>[A-Za-z_][A-Za-z0-9_]*)
+      | (?P<op>\*\*|//|[+\-*/%(),])
+    )
+""", re.VERBOSE)
+
+def tokenize(s):
+    toks = []
+    i = 0
+    n = len(s)
+    while i < n:
+        m = _TOKEN_RE.match(s, i)
+        if not m:
+            raise EvalError(f"unexpected char {s[i:i+5]!r}", i)
+        i = m.end()
+        if m.group('num') is not None:
+            toks.append(('num', float(m.group('num')), m.start()))
+        elif m.group('ident') is not None:
+            toks.append(('ident', m.group('ident'), m.start()))
+        else:
+            toks.append(('op', m.group('op'), m.start()))
+    toks.append(('eof', None, n))
+    return toks
+
+FUNCS = {
+    'sin': lambda a: math.sin(a),
+    'cos': lambda a: math.cos(a),
+    'sqrt': lambda a: math.sqrt(a),
+    'abs': lambda a: abs(a),
+    'max': lambda *a: max(a),
+    'min': lambda *a: min(a),
+    'round': lambda a: round(a),
+    'pow': lambda a, b: a ** b,
+}
+
+class Parser:
+    def __init__(self, toks, env=None):
+        self.toks = toks
+        self.pos = 0
+        self.env = env or {}
+
+    def peek(self):
+        return self.toks[self.pos]
+
+    def advance(self):
+        t = self.toks[self.pos]
+        self.pos += 1
+        return t
+
+    def expect(self, kind, val=None):
+        t = self.peek()
+        if t[0] != kind or (val is not None and t[1] != val):
+            raise EvalError(f"expected {kind} {val!r} got {t[0]} {t[1]!r}", t[2])
+        return self.advance()
+
+    # expr := term (('+'|'-') term)*
+    def parse_expr(self):
+        v = self.parse_term()
+        while self.peek()[0] == 'op' and self.peek()[1] in ('+', '-'):
+            op = self.advance()[1]
+            r = self.parse_term()
+            v = v + r if op == '+' else v - r
+        return v
+
+    def parse_term(self):
+        v = self.parse_factor()
+        while self.peek()[0] == 'op' and self.peek()[1] in ('*', '/', '//', '%'):
+            op = self.advance()[1]
+            r = self.parse_factor()
+            if op == '*':
+                v = v * r
+            elif op == '/':
+                if r == 0:
+                    raise EvalError("division by zero", self.peek()[2])
+                v = v / r
+            elif op == '//':
+                if r == 0:
+                    raise EvalError("division by zero", self.peek()[2])
+                v = v // r
+            elif op == '%':
+                if r == 0:
+                    raise EvalError("modulo by zero", self.peek()[2])
+                v = v % r
+        return v
+
+    def parse_factor(self):
+        # right-assoc **
+        base = self.parse_unary()
+        if self.peek()[0] == 'op' and self.peek()[1] == '**':
+            self.advance()
+            exp = self.parse_factor()
+            return base ** exp
+        return base
+
+    def parse_unary(self):
+        if self.peek()[0] == 'op' and self.peek()[1] in ('+', '-'):
+            op = self.advance()[1]
+            v = self.parse_unary()
+            return -v if op == '-' else v
+        return self.parse_primary()
+
+    def parse_primary(self):
+        t = self.peek()
+        if t[0] == 'num':
+            self.advance()
+            return t[1]
+        if t[0] == 'op' and t[1] == '(':
+            self.advance()
+            v = self.parse_expr()
+            self.expect('op', ')')
+            return v
+        if t[0] == 'ident':
+            name = t[1]
+            self.advance()
+            if self.peek()[0] == 'op' and self.peek()[1] == '(':
+                self.advance()
+                args = []
+                if not (self.peek()[0] == 'op' and self.peek()[1] == ')'):
+                    args.append(self.parse_expr())
+                    while self.peek()[0] == 'op' and self.peek()[1] == ',':
+                        self.advance()
+                        args.append(self.parse_expr())
+                self.expect('op', ')')
+                if name not in FUNCS:
+                    raise EvalError(f"unknown function {name}", t[2])
+                try:
+                    return FUNCS[name](*args)
+                except Exception as e:
+                    raise EvalError(f"function {name}: {e}", t[2])
+            # variable
+            if name not in self.env:
+                raise EvalError(f"unknown variable {name}", t[2])
+            return self.env[name]
+        raise EvalError(f"unexpected token {t[0]} {t[1]!r}", t[2])
+
+def evaluate(s, env=None):
+    toks = tokenize(s)
+    p = Parser(toks, env)
+    v = p.parse_expr()
+    if p.peek()[0] != 'eof':
+        raise EvalError("trailing tokens", p.peek()[2])
+    return v
+
+if __name__ == '__main__':
+    import sys
+    env = {'x': 3, 'pi': math.pi}
+    expr = ' '.join(sys.argv[1:]) or '2 + 3*4'
+    print(evaluate(expr, env))

wizard-work/beam/test_expr.py

@@ -0,0 +1,34 @@
+import math, sys
+sys.path.insert(0, '/root/.hermes/wizard-outputs/beam')
+from expr_eval import evaluate, EvalError
+
+cases = [
+    ("2+3*4", 14.0),
+    ("(2+3)*4", 20.0),
+    ("2**3**2", 512.0),  # right assoc
+    ("10 // 3", 3.0),
+    ("10 % 3", 1.0),
+    ("-5 + 2", -3.0),
+    ("2*-3", -6.0),
+    ("sqrt(16)+abs(-2)", 6.0),
+    ("max(1,5,2)", 5.0),
+    ("sin(0)", 0.0),
+    ("2**0.5", math.sqrt(2)),
+    ("1e3 + 0.5", 1000.5),
+]
+for expr, exp in cases:
+    got = evaluate(expr)
+    assert got == exp, (expr, got, exp)
+    print("OK", expr, got)
+
+# variables
+assert evaluate("x*2 + pi", {'x': 4, 'pi': math.pi}) == 8 + math.pi
+# errors
+for bad in ["2 +", "foo", "1/0", "sin(1,2)", "2 **", "abc("]:
+    try:
+        evaluate(bad)
+        print("FAIL no error:", bad)
+        sys.exit(1)
+    except EvalError as e:
+        print("ERR ok:", bad, "->", e)
+print("ALL PASS")

wizard-work/spark/test_sample.txt

@@ -1,6 +1,3 @@
-# Test file with known content
-This is a test.
-Multiple lines here.
-And a third line.
-
-Total: 4 lines, 8 words, 37 chars including newlines.
+Hello world
+This is a test file
+With three lines

wizard-work/spark/verify_wordcount_test.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+"""Verification test for wordcount.py - must match wc output exactly."""
+import subprocess
+import sys
+import tempfile
+import os
+
+# Create test file with known content
+test_content = "Hello world\nThis is a test file\nWith three lines\n"
+test_file = "/root/troupe-standup/test_verification.txt"
+with open(test_file, 'w') as f:
+    f.write(test_content)
+
+# Get wc output
+wc_result = subprocess.run(['wc', test_file], capture_output=True, text=True)
+wc_lines, wc_words, wc_chars = map(int, wc_result.stdout.strip().split()[:3])
+
+# Get wordcount.py output
+pyc_result = subprocess.run(
+    ['python3', '/root/.hermes/wizard-outputs/spark/wordcount.py', test_file],
+    capture_output=True, text=True
+)
+
+if pyc_result.returncode != 0:
+    print(f"FAIL: wordcount.py returned {pyc_result.returncode}")
+    print(pyc_result.stderr)
+    sys.exit(1)
+
+parts = pyc_result.stdout.strip().split()
+py_lines, py_words, py_chars = int(parts[0]), int(parts[1]), int(parts[2])
+
+print(f"wc:    {wc_lines} lines, {wc_words} words, {wc_chars} chars")
+print(f"wordcount.py: {py_lines} lines, {py_words} words, {py_chars} chars")
+
+if (wc_lines == py_lines and wc_words == py_words and wc_chars == py_chars):
+    print("PASS: outputs match")
+    sys.exit(0)
+else:
+    print("FAIL: outputs do not match")
+    sys.exit(1)

wizard-work/spark/wordcount.py

@@ -1,31 +1,28 @@
 #!/usr/bin/env python3
-"""Word count utility - counts words in a file and outputs JSON result."""
-import json
+"""Simple word/line/char counter for files."""
 import sys
 from pathlib import Path
 
-def word_count(filepath):
-    """Count words in file and return stats."""
-    path = Path(filepath)
-    if not path.exists():
-        return {"error": f"File not found: {filepath}"}
-    
-    content = path.read_text()
-    words = content.split()
-    lines = content.splitlines()
-    
+
+def count_file(path: str) -> dict:
+    """Return line, word, and char counts for a file (wc-style)."""
+    content = Path(path).read_text()
+    # wc -l counts newline characters, not lines
     return {
-        "file": str(path),
-        "bytes": path.stat().st_size,
-        "lines": len(lines),
-        "words": len(words),
-        "characters": len(content)
+        'lines': content.count('\n'),
+        'words': len(content.split()),
+        'chars': len(content)
     }
 
-if __name__ == "__main__":
+
+def main():
     if len(sys.argv) < 2:
-        print(json.dumps({"error": "Usage: wordcount.py <file>"}))
+        print("Usage: wordcount.py <file>", file=sys.stderr)
         sys.exit(1)
     
-    result = word_count(sys.argv[1])
-    print(json.dumps(result, indent=2))
+    result = count_file(sys.argv[1])
+    print(f"{result['lines']} {result['words']} {result['chars']} {sys.argv[1]}")
+
+
+if __name__ == '__main__':
+    main()