source/bonsai_runtime/numeric.py

"""Audited compact CPU calculator; option IDs are opaque in the runtime API."""
import ast,json,math,re
from fractions import Fraction as Q
from .readout import prompt_for

def safe_expression(text):
 text=text.strip()
 if text.startswith('```'):text=re.sub(r'^```(?:python)?\s*|\s*```$','',text).strip()
 if text in ['null','None','']:raise ValueError('abstain')
 if len(text)>1200:raise ValueError('too long')
 tree=ast.parse(text,mode='eval');assert sum(1 for _ in ast.walk(tree))<=200
 def walk(n):
  if isinstance(n,ast.Constant):
   if isinstance(n.value,bool):return n.value
   if isinstance(n.value,(int,float)) and abs(n.value)<=10**9:return Q(str(n.value))
   if isinstance(n.value,str) and len(n.value)<=100:return n.value
   raise ValueError('constant')
  if isinstance(n,(ast.List,ast.Tuple)) and len(n.elts)<=24:return [walk(x) for x in n.elts]
  if isinstance(n,ast.UnaryOp):
   v=walk(n.operand)
   if isinstance(n.op,ast.USub) and isinstance(v,Q):return -v
   if isinstance(n.op,ast.UAdd) and isinstance(v,Q):return v
   if isinstance(n.op,ast.Not) and isinstance(v,bool):return not v
  if isinstance(n,ast.BinOp):
   a,b=walk(n.left),walk(n.right)
   if not isinstance(a,Q) or not isinstance(b,Q):raise ValueError('numeric operands required')
   if isinstance(n.op,ast.Add):v=a+b
   elif isinstance(n.op,ast.Sub):v=a-b
   elif isinstance(n.op,ast.Mult):v=a*b
   elif isinstance(n.op,ast.Div):v=a/b
   elif isinstance(n.op,ast.Pow) and b.denominator==1 and abs(b)<=4:v=a**int(b)
   else:raise ValueError('operator')
   if abs(v.numerator)>10**18 or v.denominator>10**18:raise ValueError('magnitude')
   return v
  if isinstance(n,ast.Compare) and len(n.ops)==1:
   a,b=walk(n.left),walk(n.comparators[0]);op=n.ops[0]
   if isinstance(op,ast.Eq):return a==b
   if isinstance(op,ast.NotEq):return a!=b
   if isinstance(op,ast.Lt):return a<b
   if isinstance(op,ast.LtE):return a<=b
   if isinstance(op,ast.Gt):return a>b
   if isinstance(op,ast.GtE):return a>=b
  if isinstance(n,ast.BoolOp):
   v=[walk(x) for x in n.values]
   if not all(isinstance(x,bool) for x in v):raise ValueError('boolean operands required')
   return all(v) if isinstance(n.op,ast.And) else any(v)
  if isinstance(n,ast.Call) and isinstance(n.func,ast.Name) and not n.keywords:
   args=[walk(x) for x in n.args]
   if n.func.id=='order' and len(args)==2:
    edges,position=args;assert isinstance(edges,list) and isinstance(position,Q) and position.denominator==1
    assert all(isinstance(e,list) and len(e)==2 and all(isinstance(x,str) for x in e) for e in edges)
    remaining={x for e in edges for x in e};answer=[]
    while remaining:
     first=[x for x in remaining if not any(b==x and a in remaining for a,b in edges)]
     if len(first)!=1:raise ValueError('ambiguous order')
     answer+=first;remaining.remove(first[0])
    assert 1<=position<=len(answer);return answer[int(position)-1]
   if n.func.id=='first' and len(args)==2:
    conditions,values=args;assert isinstance(conditions,list) and isinstance(values,list) and len(conditions)==len(values) and all(isinstance(x,bool) for x in conditions)
    return values[conditions.index(True)]
  raise ValueError('unsupported syntax')
 return walk(tree.body)
UNITS={
 'number':('number',Q(1)), 'usd':('money',Q(1)), '$':('money',Q(1)), 'dollar':('money',Q(1)), 'dollars':('money',Q(1)), 'cent':('money',Q(1,100)), 'cents':('money',Q(1,100)),
 'km':('distance',Q(1000)), 'm':('distance',Q(1)), 'meter':('distance',Q(1)), 'meters':('distance',Q(1)), 'cm':('distance',Q(1,100)),
 'kg':('mass',Q(1000)), 'g':('mass',Q(1)), 'grams':('mass',Q(1)),
 'hours':('time',Q(3600)), 'hour':('time',Q(3600)), 'minutes':('time',Q(60)), 'minute':('time',Q(60)), 'seconds':('time',Q(1)), 'second':('time',Q(1)),
 '%':('ratio',Q(1,100)), 'percent':('ratio',Q(1,100)), 'fraction':('ratio',Q(1)),
 'l':('volume',Q(1000)), 'ml':('volume',Q(1)), 'liters':('volume',Q(1000)), 'milliliters':('volume',Q(1)), 'km/h':('speed',Q(1)), 'degrees':('temperature',Q(1))}

def quantity(text):
 text=text.strip().lower().replace(',','')
 m=re.fullmatch(r'(\$?)(-?\d+(?:\.\d+)?(?:/\d+)?)(?:[ -]*([a-z/%]+))?',text)
 if not m:return None
 unit='usd' if m[1] else m[3]
 if unit is not None and unit not in UNITS:return None
 return Q(m[2]),unit

def extended_expression(text):
 if not isinstance(text,str) or len(text)>1200:raise ValueError('invalid expression')
 tree=ast.parse(text,mode='eval')
 if sum(1 for _ in ast.walk(tree))>200:raise ValueError('expression too complex')
 class Reduce(ast.NodeTransformer):
  def visit_Call(self,node):
   self.generic_visit(node)
   if not isinstance(node.func,ast.Name) or node.func.id not in ['ceil','floor','round_to'] or node.keywords:raise ValueError('function not allowed')
   args=[safe_expression(ast.unparse(x)) for x in node.args]
   if not all(isinstance(x,Q) for x in args):raise ValueError('numeric arguments required')
   if node.func.id in ['ceil','floor'] and len(args)==1:value=Q(math.ceil(args[0]) if node.func.id=='ceil' else math.floor(args[0]))
   elif node.func.id=='round_to' and len(args)==2 and args[1].denominator==1 and 0<=args[1]<=4:
    scale=10**int(args[1]);value=Q(math.floor(abs(args[0])*scale+Q(1,2)),scale)*(1 if args[0]>=0 else -1)
   else:raise ValueError('invalid arguments')
   return ast.parse(f'({value.numerator}/{value.denominator})',mode='eval').body
  def visit_BinOp(self,node):
   self.generic_visit(node)
   if isinstance(node.op,ast.FloorDiv):
    a=safe_expression(ast.unparse(node.left));b=safe_expression(ast.unparse(node.right))
    if not isinstance(a,Q) or not isinstance(b,Q):raise ValueError('numeric division required')
    return ast.Constant(math.floor(a/b))
   return node
 tree=Reduce().visit(tree);value=safe_expression(ast.unparse(ast.fix_missing_locations(tree)))
 if not isinstance(value,Q):raise ValueError('numeric result required')
 return value

INSTRUCTION='''Return ONLY JSON with exactly "expression" and "unit". Translate the question into a short arithmetic expression using only the stated facts. Do not guess missing facts. If the requested answer cannot be calculated from the facts, return {"expression":null,"unit":null}.
Allowed arithmetic: numbers, + - * / **, //, parentheses, ceil(x), floor(x), round_to(x,digits). round_to uses nearest with half ties away from zero. Convert quantities inside the expression as necessary. Ignore identifiers and unrelated numbers. Keep arithmetic as an expression, rather than computing an answer mentally.
The unit describes the expression's output: number, USD, cents, km, m, cm, kg, g, hours, minutes, seconds, %, fraction, L, mL, km/h, degrees. Use the requested units; number is appropriate for counts or unitless answers. For Above/Below/Equal choices, calculate first quantity MINUS the comparison quantity in the same units and use unit "comparison"; the executor maps the sign. For explicitly requested rounding use round_to; for the minimum number of whole packages use ceil. Otherwise do not round.
Example: 3 groups of 8 and 2 groups of 5, combined mean -> {"expression":"(3*8+2*5)/(3+2)","unit":"number"}
Example: 2 dollars less 30 cents -> {"expression":"2-30/100","unit":"USD"}
'''

def applicable(request):
 opts=request['options'];descs={d.strip().lower() for _,d in opts}
 return sum(quantity(d) is not None for _,d in opts)>=2 or {'above','below','equal'}<=descs

def decode(text,options):
 data=json.loads(text)
 if not isinstance(data,dict) or set(data)!={'expression','unit'}:raise ValueError('invalid schema')
 if data['expression'] is None:raise ValueError('abstain')
 if not isinstance(data['unit'],str):raise ValueError('invalid unit')
 unit=data['unit'].strip().lower();value=extended_expression(data['expression']);matches=[]
 if unit=='comparison':
  word='above' if value>0 else 'below' if value<0 else 'equal';matches=[k for k,d in options if d.strip().lower()==word]
 else:
  if unit not in UNITS:raise ValueError('unknown unit')
  dim,factor=UNITS[unit]
  for k,d in options:
   number=quantity(d)
   if number is None:continue
   n,u=number
   if u is None:match=n==value
   else:od,of=UNITS[u];match=od==dim and n*of==value*factor
   if match:matches.append(k)
 if len(matches)!=1:raise ValueError('no unique unit-consistent match')
 return matches[0],data

def compact_unit(request):
 descs={d.strip().lower() for _,d in request['options']}
 if {'above','below','equal'}<=descs:return 'comparison'
 found=[quantity(d) for k,d in request['options']]
 units={x[1] for x in found if x is not None and x[1] is not None}
 return next(iter(units)) if len(units)==1 else 'number' if not units else None

def compact_prompt(request):
 unit=compact_unit(request)
 instruction=f'''Return only one arithmetic expression, not an answer letter or explanation. Calculate the requested quantity from the supplied facts; ignore identifiers and irrelevant numbers. Output unit: {unit}. If facts are insufficient, return null. Allowed: numbers, + - * / ** //, parentheses, ceil(x), floor(x), round_to(x,digits). round_to rounds nearest, half ties away from zero. Convert mixed units before arithmetic. Apply rounding only when requested; use ceil for a minimum whole-package purchase. Keep the calculation as an expression instead of doing it mentally.'''
 if unit=='comparison':instruction+=' Compute the first quantity MINUS the comparison quantity, after converting to matching units. The sign determines Above/Below/Equal.'
 return instruction+'\n'+prompt_for(request['state'],request['question'],request['options']).replace('Answer with the letter of the best option only.','Return only the arithmetic expression, or null.')