let domains author their own pass-2 estimator as formulas, not Python

Pass 2 previously only knew one estimator: a hardcoded physics model that
matches dimensions literally named platform/actuator/energy_storage. Any
domain outside that shape (e.g. archery) got all-zero estimates and failed
every combo. Domains can now declare free variables and per-metric formulas
as data instead; a safe AST-based evaluator (engine/formula.py, no eval())
resolves declared entity properties via dep(key, constraint_type) and
generalizes the existing hand-nested mass-budget search into an N-variable
recursive optimizer. Fully additive -- the legacy platform/actuator/
energy_storage path is untouched and still runs unchanged for every domain
that declares no formulas.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-16 21:28:15 -05:00
parent 3795a7e826
commit 3429bce8d0
15 changed files with 1064 additions and 9 deletions

View File

@@ -9,7 +9,7 @@ from datetime import datetime, timezone
from typing import Sequence
from physcom.models.entity import Dependency, Entity
from physcom.models.domain import Domain, DomainConstraint, MetricBound
from physcom.models.domain import Domain, DomainConstraint, FreeVariable, MetricBound, MetricFormula
from physcom.models.combination import Combination
@@ -286,6 +286,10 @@ class Repository:
"INSERT OR IGNORE INTO domain_constraints (domain_id, key, value) VALUES (?, ?, ?)",
(domain.id, dc.key, val),
)
for fv in domain.free_variables:
self.add_free_variable(domain.id, fv, commit=False)
for mf in domain.metric_formulas:
self.add_metric_formula(domain.id, mf, commit=False)
self.conn.commit()
return domain
@@ -299,6 +303,31 @@ class Repository:
by_key.setdefault(r["key"], []).append(r["value"])
return [DomainConstraint(key=k, allowed_values=v) for k, v in by_key.items()]
def _load_free_variables(self, domain_id: int) -> list[FreeVariable]:
rows = self.conn.execute(
"""SELECT id, name, sort_order, floor_formula, ceiling_formula
FROM domain_free_variables WHERE domain_id = ? ORDER BY sort_order""",
(domain_id,),
).fetchall()
return [
FreeVariable(
id=r["id"], name=r["name"], sort_order=r["sort_order"],
floor_formula=r["floor_formula"], ceiling_formula=r["ceiling_formula"],
)
for r in rows
]
def _load_metric_formulas(self, domain_id: int) -> list[MetricFormula]:
rows = self.conn.execute(
"""SELECT id, metric_name, formula
FROM domain_metric_formulas WHERE domain_id = ? ORDER BY metric_name""",
(domain_id,),
).fetchall()
return [
MetricFormula(id=r["id"], metric_name=r["metric_name"], formula=r["formula"])
for r in rows
]
def _load_domain(self, where: str, param: str | int) -> Domain | None:
row = self.conn.execute(f"SELECT * FROM domains WHERE {where} = ?", (param,)).fetchone()
if not row:
@@ -325,6 +354,8 @@ class Repository:
for w in weights
],
constraints=self._load_domain_constraints(row["id"]),
free_variables=self._load_free_variables(row["id"]),
metric_formulas=self._load_metric_formulas(row["id"]),
)
def get_domain(self, name: str) -> Domain | None:
@@ -376,12 +407,63 @@ class Repository:
)
self.conn.commit()
# ── Free variables & metric formulas ──────────────────────────
def add_free_variable(self, domain_id: int, fv: FreeVariable, commit: bool = True) -> FreeVariable:
cur = self.conn.execute(
"""INSERT INTO domain_free_variables
(domain_id, name, sort_order, floor_formula, ceiling_formula)
VALUES (?, ?, ?, ?, ?)""",
(domain_id, fv.name, fv.sort_order, fv.floor_formula, fv.ceiling_formula),
)
fv.id = cur.lastrowid
if commit:
self.conn.commit()
return fv
def update_free_variable(self, fv_id: int, fv: FreeVariable) -> None:
self.conn.execute(
"""UPDATE domain_free_variables
SET name = ?, sort_order = ?, floor_formula = ?, ceiling_formula = ?
WHERE id = ?""",
(fv.name, fv.sort_order, fv.floor_formula, fv.ceiling_formula, fv_id),
)
self.conn.commit()
def delete_free_variable(self, fv_id: int) -> None:
self.conn.execute("DELETE FROM domain_free_variables WHERE id = ?", (fv_id,))
self.conn.commit()
def add_metric_formula(self, domain_id: int, mf: MetricFormula, commit: bool = True) -> MetricFormula:
cur = self.conn.execute(
"""INSERT OR REPLACE INTO domain_metric_formulas (domain_id, metric_name, formula)
VALUES (?, ?, ?)""",
(domain_id, mf.metric_name, mf.formula),
)
mf.id = cur.lastrowid
if commit:
self.conn.commit()
return mf
def update_metric_formula(self, mf_id: int, mf: MetricFormula) -> None:
self.conn.execute(
"UPDATE domain_metric_formulas SET metric_name = ?, formula = ? WHERE id = ?",
(mf.metric_name, mf.formula, mf_id),
)
self.conn.commit()
def delete_metric_formula(self, mf_id: int) -> None:
self.conn.execute("DELETE FROM domain_metric_formulas WHERE id = ?", (mf_id,))
self.conn.commit()
def delete_domain(self, domain_id: int) -> None:
self.conn.execute("DELETE FROM pipeline_runs WHERE domain_id = ?", (domain_id,))
self.conn.execute("DELETE FROM combination_results WHERE domain_id = ?", (domain_id,))
self.conn.execute("DELETE FROM combination_scores WHERE domain_id = ?", (domain_id,))
self.conn.execute("DELETE FROM domain_metric_weights WHERE domain_id = ?", (domain_id,))
self.conn.execute("DELETE FROM domain_constraints WHERE domain_id = ?", (domain_id,))
self.conn.execute("DELETE FROM domain_free_variables WHERE domain_id = ?", (domain_id,))
self.conn.execute("DELETE FROM domain_metric_formulas WHERE domain_id = ?", (domain_id,))
self.conn.execute("DELETE FROM domains WHERE id = ?", (domain_id,))
self.conn.commit()
@@ -906,6 +988,8 @@ class Repository:
self.conn.execute("DELETE FROM entities")
self.conn.execute("DELETE FROM domain_metric_weights")
self.conn.execute("DELETE FROM domain_constraints")
self.conn.execute("DELETE FROM domain_free_variables")
self.conn.execute("DELETE FROM domain_metric_formulas")
self.conn.execute("DELETE FROM domains")
self.conn.execute("DELETE FROM metrics")
self.conn.execute("DELETE FROM dimensions")

View File

@@ -120,6 +120,24 @@ CREATE TABLE IF NOT EXISTS domain_constraints (
UNIQUE(domain_id, key, value)
);
CREATE TABLE IF NOT EXISTS domain_free_variables (
id INTEGER PRIMARY KEY AUTOINCREMENT,
domain_id INTEGER NOT NULL REFERENCES domains(id),
name TEXT NOT NULL,
sort_order INTEGER NOT NULL,
floor_formula TEXT NOT NULL,
ceiling_formula TEXT NOT NULL,
UNIQUE(domain_id, name)
);
CREATE TABLE IF NOT EXISTS domain_metric_formulas (
id INTEGER PRIMARY KEY AUTOINCREMENT,
domain_id INTEGER NOT NULL REFERENCES domains(id),
metric_name TEXT NOT NULL,
formula TEXT NOT NULL,
UNIQUE(domain_id, metric_name)
);
CREATE INDEX IF NOT EXISTS idx_deps_entity ON dependencies(entity_id);
CREATE INDEX IF NOT EXISTS idx_deps_category_key ON dependencies(category, key);
CREATE INDEX IF NOT EXISTS idx_combo_status ON combinations(status);

View File

@@ -60,6 +60,40 @@ KEY_AGGREGATION: dict[str, str] = {
OVERRUN_TOLERANCE: float = 0.10
def aggregate_dependency_value(
combination: Combination,
key: str,
constraint_type: str,
key_aggregation: dict[str, str] | None = None,
) -> float | None:
"""Collapse every numeric dependency matching (key, constraint_type)
across a combination's entities into one system-level number: summed for
extensive keys (KEY_AGGREGATION says "sum", e.g. mass/footprint --
independent components sharing one physical vehicle), otherwise the
strongest single value wins (today's default pairwise behavior). Returns
None if no entity declares a matching numeric dependency. Shared by
ConstraintResolver._check_provides_vs_range and the formula evaluator's
injected dep() builtin (see engine/formula.py, engine/pipeline.py) so
there's one implementation of "how do these entities' declared numbers
combine," not two.
"""
key_aggregation = KEY_AGGREGATION if key_aggregation is None else key_aggregation
values: list[float] = []
for entity in combination.entities:
for dep in entity.dependencies:
if dep.key != key or dep.constraint_type != constraint_type:
continue
try:
values.append(float(dep.value))
except (ValueError, TypeError):
continue
if not values:
return None
if key_aggregation.get(key) == "sum":
return sum(values)
return max(values)
@dataclass
class ConstraintResult:
"""Outcome of constraint resolution for a combination."""
@@ -250,11 +284,13 @@ class ConstraintResolver:
required.setdefault(dep.key, []).append((entity.name, val))
for key in set(provided) & set(required):
prov_val = aggregate_dependency_value(
combination, key, "provides", self.key_aggregation
)
if self.key_aggregation.get(key) == "sum":
prov_name = " + ".join(name for name, _ in provided[key])
prov_val = sum(val for _, val in provided[key])
else:
prov_name, prov_val = max(provided[key], key=lambda t: t[1])
prov_name, _ = max(provided[key], key=lambda t: t[1])
for req_name, req_val in required[key]:
if prov_val < req_val * self.deficit_threshold:

View File

@@ -0,0 +1,139 @@
"""Safe arithmetic expression language for domain-authored estimator formulas.
No eval()/exec() anywhere -- compile_formula validates every AST node against
a fixed whitelist (arithmetic, numeric/string constants, name lookups, calls
to an explicitly supplied function table) before evaluate_formula ever walks
it, so a formula can express "how much drawback_force does this combo
produce" but never anything with attribute/subscript/import access.
"""
from __future__ import annotations
import ast
import math
import operator
from dataclasses import dataclass
from typing import Callable
class FormulaError(Exception):
"""Raised for invalid formula syntax/structure or a failed evaluation."""
_ALLOWED_NODES = (
ast.Expression, ast.BinOp, ast.UnaryOp, ast.Constant, ast.Name, ast.Load,
ast.Call, ast.keyword,
ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Pow, ast.USub, ast.UAdd,
)
_BINOPS: dict[type, Callable[[float, float], float]] = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
}
_UNARYOPS: dict[type, Callable[[float], float]] = {
ast.USub: operator.neg,
ast.UAdd: operator.pos,
}
DEFAULT_FUNCTIONS: dict[str, Callable[..., float]] = {
"min": min,
"max": max,
"abs": abs,
"sqrt": math.sqrt,
"log": math.log,
"log1p": math.log1p,
"exp": math.exp,
}
@dataclass(frozen=True)
class CompiledFormula:
source: str
_tree: ast.Expression
def _validate(tree: ast.AST) -> None:
for node in ast.walk(tree):
if not isinstance(node, _ALLOWED_NODES):
raise FormulaError(
f"disallowed expression element: {type(node).__name__}"
)
if isinstance(node, ast.Constant):
if isinstance(node.value, bool) or not isinstance(node.value, (int, float, str)):
raise FormulaError(
f"disallowed constant type: {type(node.value).__name__}"
)
if isinstance(node, ast.Name) and node.id.startswith("__"):
raise FormulaError(f"disallowed name: {node.id}")
if isinstance(node, ast.Call) and not isinstance(node.func, ast.Name):
raise FormulaError("only direct function calls are allowed")
def compile_formula(source: str) -> CompiledFormula:
try:
tree = ast.parse(source, mode="eval")
except SyntaxError as exc:
raise FormulaError(f"invalid syntax in '{source}': {exc}") from exc
_validate(tree)
return CompiledFormula(source=source, _tree=tree)
def _eval(node: ast.AST, variables: dict[str, float], functions: dict[str, Callable]):
if isinstance(node, ast.Expression):
return _eval(node.body, variables, functions)
if isinstance(node, ast.Constant):
# Numeric constants are cast to float (not left as int) so a formula
# like a**b**c can't build an arbitrary-precision giant int before
# ever raising -- float exponentiation overflows to inf/OverflowError
# quickly instead. String constants (dep() key/constraint_type args)
# pass through unchanged.
return node.value if isinstance(node.value, str) else float(node.value)
if isinstance(node, ast.Name):
if node.id not in variables:
raise FormulaError(f"unknown variable '{node.id}'")
return variables[node.id]
if isinstance(node, ast.BinOp):
op = _BINOPS.get(type(node.op))
if op is None:
raise FormulaError(f"unsupported operator: {type(node.op).__name__}")
return op(
_eval(node.left, variables, functions),
_eval(node.right, variables, functions),
)
if isinstance(node, ast.UnaryOp):
op = _UNARYOPS.get(type(node.op))
if op is None:
raise FormulaError(f"unsupported operator: {type(node.op).__name__}")
return op(_eval(node.operand, variables, functions))
if isinstance(node, ast.Call):
fname = node.func.id # validated as ast.Name by _validate
func = functions.get(fname)
if func is None:
raise FormulaError(f"unknown function '{fname}'")
args = [_eval(a, variables, functions) for a in node.args]
kwargs = {kw.arg: _eval(kw.value, variables, functions) for kw in node.keywords}
return func(*args, **kwargs)
raise FormulaError(f"unsupported expression element: {type(node).__name__}")
def evaluate_formula(
compiled: CompiledFormula,
variables: dict[str, float],
functions: dict[str, Callable] | None = None,
) -> float:
effective_functions = {**DEFAULT_FUNCTIONS, **(functions or {})}
try:
result = _eval(compiled._tree, variables, effective_functions)
except FormulaError:
raise
except (TypeError, ValueError, ArithmeticError) as exc:
raise FormulaError(f"error evaluating '{compiled.source}': {exc}") from exc
if not isinstance(result, (int, float)) or isinstance(result, bool):
raise FormulaError(
f"formula '{compiled.source}' did not evaluate to a number"
)
return float(result)

View File

@@ -674,7 +674,7 @@ class Pipeline:
result.pass2_estimated += 1
return
raw_metrics, feasible = self._estimate_physics(combo, domain.metric_bounds)
raw_metrics, feasible = self._estimate_physics(combo, domain)
if not feasible:
# No platform mass within its own declared ceiling could
@@ -1324,11 +1324,19 @@ class Pipeline:
return best_a, best_s, best_score
def _estimate_physics(
self, combo: Combination, metric_bounds: list[MetricBound]
self, combo: Combination, domain: Domain
) -> tuple[dict[str, float], bool]:
"""Deterministic physics-based estimation from declared entity
attributes (no LLM) -- pass 2's estimator.
Domains that declare their own metric_formulas (see
_estimate_via_formulas) are estimated entirely by those
domain-authored formulas instead -- this method's hardcoded
platform/actuator/energy_storage physics model below is additive,
not the only path: it's untouched and still runs unchanged for
every domain that declares no formulas (every existing transport
domain today).
power_density, speed, range_fuel, cost_efficiency, and
cargo_capacity/cargo_capacity_kg are all computed from the
platform's declared mass envelope treated as a combo-wide budget,
@@ -1356,6 +1364,10 @@ class Pipeline:
whose absolute mass tramples its own platform's declared ceiling
can still produce perfectly plausible-looking per-kg ratios.
"""
if domain.metric_formulas:
return self._estimate_via_formulas(combo, domain)
metric_bounds = domain.metric_bounds
metric_names = [mb.metric_name for mb in metric_bounds]
units_by_name = {mb.metric_name: mb.unit for mb in metric_bounds}
bounds_by_name = {mb.metric_name: mb for mb in metric_bounds}
@@ -1427,6 +1439,169 @@ class Pipeline:
return raw, feasible
def _estimate_via_formulas(
self, combo: Combination, domain: Domain
) -> tuple[dict[str, float], bool]:
"""Generic estimator for domains that declare their own
metric_formulas/free_variables instead of matching the
platform/actuator/energy_storage shape _estimate_physics assumes.
Every formula (free-variable floor/ceiling, and each metric) is
evaluated through the same safe engine.formula evaluator, given two
kinds of names to resolve: dep(key, constraint_type="provides",
agg=None) -- a declared entity property, aggregated across the
combo's entities via constraint_resolver.aggregate_dependency_value
(the exact sum-vs-max rule ConstraintResolver itself already
applies to mass/footprint) -- and any free variable already chosen
earlier in its declared sort_order, as a bare name. Free variables
are resolved by _search_free_variables, a generalization of
_search_best_allocation/_decide_masses's hand-nested 2-3-variable
search to an arbitrary domain-declared list; with zero free
variables (most new domains -- nothing needs sizing) it degenerates
to a single direct formula evaluation, no search at all.
An invalid formula (bad syntax, unknown name, division by zero for
this combo's values) degrades that one metric to 0.0 / that one free
variable to being skipped, rather than failing the whole combo --
symmetric with how a metric_bounds entry with no matching raw value
already defaults to 0.0 elsewhere in this pipeline.
"""
from physcom.engine.constraint_resolver import aggregate_dependency_value
from physcom.engine.formula import FormulaError, compile_formula, evaluate_formula
def dep(key: str, constraint_type: str = "provides", agg: str | None = None) -> float:
key_aggregation = None if agg is None else {key: agg}
value = aggregate_dependency_value(combo, key, constraint_type, key_aggregation)
return value if value is not None else 0.0
functions = {"dep": dep}
compiled_metrics: dict[str, object] = {}
for mf in domain.metric_formulas:
try:
compiled_metrics[mf.metric_name] = compile_formula(mf.formula)
except FormulaError:
compiled_metrics[mf.metric_name] = None
compiled_vars: list[tuple[str, object, object]] = []
for fv in sorted(domain.free_variables, key=lambda v: v.sort_order):
try:
compiled_vars.append((
fv.name, compile_formula(fv.floor_formula), compile_formula(fv.ceiling_formula),
))
except FormulaError:
continue
bounds_by_name = {mb.metric_name: mb for mb in domain.metric_bounds}
def evaluate_metrics(resolved: dict[str, float]) -> dict[str, float]:
raw: dict[str, float] = {}
for name, compiled in compiled_metrics.items():
if compiled is None:
raw[name] = 0.0
continue
try:
raw[name] = evaluate_formula(compiled, resolved, functions)
except FormulaError:
raw[name] = 0.0
return raw
def objective(resolved: dict[str, float]) -> float:
raw = evaluate_metrics(resolved)
scores, weights = [], []
for mb in bounds_by_name.values():
val = raw.get(mb.metric_name)
if val is None:
continue
n = normalize(val, mb.norm_min, mb.norm_max)
if mb.lower_is_better:
n = 1.0 - n
scores.append(n)
weights.append(mb.weight)
return composite_score(scores, weights)
resolved, _score, feasible = self._search_free_variables(
compiled_vars, {}, functions, objective
)
return evaluate_metrics(resolved), feasible
def _search_free_variables(
self,
var_specs: list[tuple[str, object, object]],
resolved: dict[str, float],
functions: dict,
objective,
grid: int = 10,
rounds: int = 5,
) -> tuple[dict[str, float], float, bool]:
"""Recursive generalization of _search_best_allocation/_decide_masses's
hand-nested outer-platform/inner-(actuator,storage) search to an
arbitrary ordered list of free variables: resolve var_specs[0]'s
floor/ceiling against what's already been resolved plus dep(...),
coarse-to-fine grid-search it, and recurse into the rest for each
candidate -- exactly the same nesting the hardcoded 3-variable
search already does by hand, just generalized to N. grid/rounds
shrink with remaining depth (cost is grid**depth * rounds**depth)
for the same reason the hardcoded search's outer loop already uses
a cheaper inner search per candidate; tuned for a handful of free
variables (transport's own 3 is the reference point), not domains
declaring dozens. Returns (resolved, score, feasible); feasible is
False only when no candidate at some depth has floor <= ceiling.
"""
from physcom.engine.formula import FormulaError, evaluate_formula
if not var_specs:
return dict(resolved), objective(resolved), True
name, floor_f, ceiling_f = var_specs[0]
rest = var_specs[1:]
try:
floor = evaluate_formula(floor_f, resolved, functions)
ceiling = evaluate_formula(ceiling_f, resolved, functions)
except FormulaError:
result = dict(resolved)
result[name] = 0.0
return result, -1.0, False
if ceiling < floor:
result = dict(resolved)
result[name] = floor
return result, -1.0, False
inner_grid = max(4, grid - 2 * len(rest))
inner_rounds = max(2, rounds - len(rest))
if ceiling == floor:
candidate = dict(resolved)
candidate[name] = floor
return self._search_free_variables(
rest, candidate, functions, objective, inner_grid, inner_rounds
)
win_lo, win_hi = floor, ceiling
best_resolved, best_score, best_feasible = None, -1.0, False
for _round in range(rounds):
for i in range(grid + 1):
v = win_lo + (win_hi - win_lo) * i / grid
candidate = dict(resolved)
candidate[name] = v
r_resolved, r_score, r_feasible = self._search_free_variables(
rest, candidate, functions, objective, inner_grid, inner_rounds
)
if r_score > best_score:
best_resolved, best_score, best_feasible = r_resolved, r_score, r_feasible
if best_resolved is None:
break
span = max((win_hi - win_lo) / grid * 2, 1e-9)
center = best_resolved[name]
win_lo, win_hi = max(floor, center - span), min(ceiling, center + span)
if best_resolved is None:
result = dict(resolved)
result[name] = floor
return result, -1.0, False
return best_resolved, best_score, best_feasible
def evaluate_allocation(
self,
combo: Combination,

View File

@@ -26,6 +26,32 @@ class DomainConstraint:
allowed_values: list[str] = field(default_factory=list) # e.g. ["ground", "air"]
@dataclass
class FreeVariable:
"""A domain-declared quantity pass 2's estimator searches to maximize
the composite score (see Pipeline._estimate_via_formulas), e.g.
"actuator_mass". floor_formula/ceiling_formula are evaluated per-combo
and may reference dep(...) and any free variable declared at a lower
sort_order (mirrors the outer/inner nesting the built-in transport
physics model already does by hand)."""
name: str
floor_formula: str
ceiling_formula: str
sort_order: int = 0
id: int | None = None
@dataclass
class MetricFormula:
"""A domain-declared formula computing one metric's raw value, evaluated
against dep(...) lookups and the domain's resolved free variables."""
metric_name: str
formula: str
id: int | None = None
@dataclass
class Domain:
"""A context frame that defines what 'good' means (e.g., urban_commuting)."""
@@ -34,4 +60,6 @@ class Domain:
description: str = ""
metric_bounds: list[MetricBound] = field(default_factory=list)
constraints: list[DomainConstraint] = field(default_factory=list)
free_variables: list[FreeVariable] = field(default_factory=list)
metric_formulas: list[MetricFormula] = field(default_factory=list)
id: int | None = None

View File

@@ -6,7 +6,7 @@ from datetime import datetime, timezone
from physcom.db.repository import Repository
from physcom.models.entity import Entity, Dependency
from physcom.models.domain import Domain, DomainConstraint, MetricBound
from physcom.models.domain import Domain, DomainConstraint, FreeVariable, MetricBound, MetricFormula
from physcom.models.combination import Combination
@@ -70,11 +70,27 @@ def export_snapshot(repo: Repository) -> dict:
"key": dc.key,
"allowed_values": dc.allowed_values,
})
fvs = []
for fv in d.free_variables:
fvs.append({
"name": fv.name,
"sort_order": fv.sort_order,
"floor_formula": fv.floor_formula,
"ceiling_formula": fv.ceiling_formula,
})
mfs = []
for mf in d.metric_formulas:
mfs.append({
"metric_name": mf.metric_name,
"formula": mf.formula,
})
domain_list.append({
"name": d.name,
"description": d.description,
"metric_bounds": mbs,
"constraints": dcs,
"free_variables": fvs,
"metric_formulas": mfs,
})
# Export combinations
@@ -207,11 +223,26 @@ def import_snapshot(repo: Repository, data: dict, *, clear: bool = False) -> dic
)
for dc in d_data.get("constraints", [])
]
fvs = [
FreeVariable(
name=fv["name"],
floor_formula=fv["floor_formula"],
ceiling_formula=fv["ceiling_formula"],
sort_order=fv.get("sort_order", 0),
)
for fv in d_data.get("free_variables", [])
]
mfs = [
MetricFormula(metric_name=mf["metric_name"], formula=mf["formula"])
for mf in d_data.get("metric_formulas", [])
]
domain = Domain(
name=d_data["name"],
description=d_data.get("description", ""),
metric_bounds=mbs,
constraints=dcs,
free_variables=fvs,
metric_formulas=mfs,
)
repo.add_domain(domain)
counts["domains"] += 1

View File

@@ -4,7 +4,7 @@ from __future__ import annotations
from flask import Blueprint, flash, redirect, render_template, request, url_for
from physcom.models.domain import Domain, MetricBound
from physcom.models.domain import Domain, FreeVariable, MetricBound, MetricFormula
from physcom_web.app import get_repo
bp = Blueprint("domains", __name__, url_prefix="/domains")
@@ -117,3 +117,96 @@ def metric_delete(domain_id: int, metric_id: int):
flash("Metric removed.", "success")
domain = repo.get_domain_by_id(domain_id)
return render_template("domains/_metrics_table.html", domain=domain)
# ── Free variable CRUD (HTMX partials) ───────────────────────
@bp.route("/<int:domain_id>/free-vars/add", methods=["POST"])
def free_var_add(domain_id: int):
repo = get_repo()
name = request.form["name"].strip()
floor_formula = request.form.get("floor_formula", "").strip()
ceiling_formula = request.form.get("ceiling_formula", "").strip()
try:
sort_order = int(request.form.get("sort_order", "0"))
except ValueError:
sort_order = 0
if not name or not floor_formula or not ceiling_formula:
flash("Name, floor formula, and ceiling formula are required.", "error")
else:
fv = FreeVariable(
name=name, floor_formula=floor_formula, ceiling_formula=ceiling_formula,
sort_order=sort_order,
)
repo.add_free_variable(domain_id, fv)
flash(f"Free variable '{name}' added.", "success")
domain = repo.get_domain_by_id(domain_id)
return render_template("domains/_free_vars_table.html", domain=domain)
@bp.route("/<int:domain_id>/free-vars/<int:fv_id>/edit", methods=["POST"])
def free_var_edit(domain_id: int, fv_id: int):
repo = get_repo()
try:
sort_order = int(request.form.get("sort_order", "0"))
except ValueError:
sort_order = 0
fv = FreeVariable(
name=request.form["name"].strip(),
floor_formula=request.form.get("floor_formula", "").strip(),
ceiling_formula=request.form.get("ceiling_formula", "").strip(),
sort_order=sort_order,
)
repo.update_free_variable(fv_id, fv)
flash("Free variable updated.", "success")
domain = repo.get_domain_by_id(domain_id)
return render_template("domains/_free_vars_table.html", domain=domain)
@bp.route("/<int:domain_id>/free-vars/<int:fv_id>/delete", methods=["POST"])
def free_var_delete(domain_id: int, fv_id: int):
repo = get_repo()
repo.delete_free_variable(fv_id)
flash("Free variable removed.", "success")
domain = repo.get_domain_by_id(domain_id)
return render_template("domains/_free_vars_table.html", domain=domain)
# ── Metric formula CRUD (HTMX partials) ──────────────────────
@bp.route("/<int:domain_id>/formulas/add", methods=["POST"])
def formula_add(domain_id: int):
repo = get_repo()
metric_name = request.form["metric_name"].strip()
formula = request.form.get("formula", "").strip()
if not metric_name or not formula:
flash("Metric name and formula are required.", "error")
else:
repo.add_metric_formula(domain_id, MetricFormula(metric_name=metric_name, formula=formula))
flash(f"Formula for '{metric_name}' added.", "success")
domain = repo.get_domain_by_id(domain_id)
return render_template("domains/_formulas_table.html", domain=domain)
@bp.route("/<int:domain_id>/formulas/<int:formula_id>/edit", methods=["POST"])
def formula_edit(domain_id: int, formula_id: int):
repo = get_repo()
mf = MetricFormula(
metric_name=request.form["metric_name"].strip(),
formula=request.form.get("formula", "").strip(),
)
repo.update_metric_formula(formula_id, mf)
flash("Formula updated.", "success")
domain = repo.get_domain_by_id(domain_id)
return render_template("domains/_formulas_table.html", domain=domain)
@bp.route("/<int:domain_id>/formulas/<int:formula_id>/delete", methods=["POST"])
def formula_delete(domain_id: int, formula_id: int):
repo = get_repo()
repo.delete_metric_formula(formula_id)
flash("Formula removed.", "success")
domain = repo.get_domain_by_id(domain_id)
return render_template("domains/_formulas_table.html", domain=domain)

View File

@@ -0,0 +1,56 @@
<table id="formulas-table">
<thead>
<tr>
<th>Metric</th>
<th>Formula</th>
<th></th>
</tr>
</thead>
<tbody>
{% for mf in domain.metric_formulas %}
<tr>
<td>{{ mf.metric_name }}</td>
<td><code>{{ mf.formula }}</code></td>
<td class="actions">
<button class="btn btn-sm"
onclick="this.closest('tr').nextElementSibling.style.display='table-row'; this.closest('tr').style.display='none'">
Edit
</button>
<form method="post"
hx-post="{{ url_for('domains.formula_delete', domain_id=domain.id, formula_id=mf.id) }}"
hx-target="#formulas-section" hx-swap="innerHTML"
class="inline-form">
<button type="submit" class="btn btn-sm btn-danger">Del</button>
</form>
</td>
</tr>
<tr class="edit-row" style="display:none">
<form method="post"
hx-post="{{ url_for('domains.formula_edit', domain_id=domain.id, formula_id=mf.id) }}"
hx-target="#formulas-section" hx-swap="innerHTML">
<td><input name="metric_name" value="{{ mf.metric_name }}" required></td>
<td><input name="formula" value="{{ mf.formula }}" required></td>
<td>
<button type="submit" class="btn btn-sm btn-primary">Save</button>
<button type="button" class="btn btn-sm"
onclick="this.closest('tr').style.display='none'; this.closest('tr').previousElementSibling.style.display=''">
Cancel
</button>
</td>
</form>
</tr>
{% endfor %}
</tbody>
</table>
<h3>Add Formula</h3>
<form method="post"
hx-post="{{ url_for('domains.formula_add', domain_id=domain.id) }}"
hx-target="#formulas-section" hx-swap="innerHTML"
class="dep-add-form">
<div class="form-row">
<input name="metric_name" placeholder="metric name" required>
<input name="formula" placeholder='formula, e.g. draw_weight_chosen * 2' required>
<button type="submit" class="btn btn-primary">Add</button>
</div>
</form>

View File

@@ -0,0 +1,64 @@
<table id="free-vars-table">
<thead>
<tr>
<th>Name</th>
<th>Order</th>
<th>Floor formula</th>
<th>Ceiling formula</th>
<th></th>
</tr>
</thead>
<tbody>
{% for fv in domain.free_variables %}
<tr>
<td>{{ fv.name }}</td>
<td>{{ fv.sort_order }}</td>
<td><code>{{ fv.floor_formula }}</code></td>
<td><code>{{ fv.ceiling_formula }}</code></td>
<td class="actions">
<button class="btn btn-sm"
onclick="this.closest('tr').nextElementSibling.style.display='table-row'; this.closest('tr').style.display='none'">
Edit
</button>
<form method="post"
hx-post="{{ url_for('domains.free_var_delete', domain_id=domain.id, fv_id=fv.id) }}"
hx-target="#free-vars-section" hx-swap="innerHTML"
class="inline-form">
<button type="submit" class="btn btn-sm btn-danger">Del</button>
</form>
</td>
</tr>
<tr class="edit-row" style="display:none">
<form method="post"
hx-post="{{ url_for('domains.free_var_edit', domain_id=domain.id, fv_id=fv.id) }}"
hx-target="#free-vars-section" hx-swap="innerHTML">
<td><input name="name" value="{{ fv.name }}" required></td>
<td><input name="sort_order" type="number" step="1" value="{{ fv.sort_order }}"></td>
<td><input name="floor_formula" value="{{ fv.floor_formula }}" required></td>
<td><input name="ceiling_formula" value="{{ fv.ceiling_formula }}" required></td>
<td>
<button type="submit" class="btn btn-sm btn-primary">Save</button>
<button type="button" class="btn btn-sm"
onclick="this.closest('tr').style.display='none'; this.closest('tr').previousElementSibling.style.display=''">
Cancel
</button>
</td>
</form>
</tr>
{% endfor %}
</tbody>
</table>
<h3>Add Free Variable</h3>
<form method="post"
hx-post="{{ url_for('domains.free_var_add', domain_id=domain.id) }}"
hx-target="#free-vars-section" hx-swap="innerHTML"
class="dep-add-form">
<div class="form-row">
<input name="name" placeholder="name, e.g. draw_weight_chosen" required>
<input name="sort_order" type="number" step="1" placeholder="order" value="0">
<input name="floor_formula" placeholder='floor formula, e.g. dep("draw_weight", "range_min")' required>
<input name="ceiling_formula" placeholder='ceiling formula, e.g. dep("draw_weight", "range_max")' required>
<button type="submit" class="btn btn-primary">Add</button>
</div>
</form>

View File

@@ -33,4 +33,18 @@
<div id="metrics-section">
{% include "domains/_metrics_table.html" %}
</div>
<h2>Free Variables</h2>
<p class="hint">Quantities pass 2's estimator searches to maximize this domain's composite score. Leave empty for domains where nothing needs sizing.</p>
<div id="free-vars-section">
{% include "domains/_free_vars_table.html" %}
</div>
<h2>Metric Formulas</h2>
<p class="hint">How each metric's raw value is computed from declared entity properties (via <code>dep(key, constraint_type="provides")</code>) and any free variable above. If this domain declares any formulas, pass 2 uses them instead of the built-in vehicle physics model.</p>
<div id="formulas-section">
{% include "domains/_formulas_table.html" %}
</div>
{% endblock %}

108
tests/test_formula.py Normal file
View File

@@ -0,0 +1,108 @@
"""Tests for the safe formula evaluator."""
import math
import pytest
from physcom.engine.formula import FormulaError, compile_formula, evaluate_formula
class TestArithmetic:
def test_constant(self):
assert evaluate_formula(compile_formula("42"), {}) == 42.0
def test_basic_ops(self):
assert evaluate_formula(compile_formula("2 + 3 * 4"), {}) == 14.0
assert evaluate_formula(compile_formula("(2 + 3) * 4"), {}) == 20.0
assert evaluate_formula(compile_formula("10 / 4"), {}) == 2.5
assert evaluate_formula(compile_formula("2 ** 3"), {}) == 8.0
def test_unary_minus(self):
assert evaluate_formula(compile_formula("-5 + 2"), {}) == -3.0
def test_variable_lookup(self):
result = evaluate_formula(compile_formula("mass * 2"), {"mass": 3.0})
assert result == 6.0
def test_unknown_variable_raises(self):
with pytest.raises(FormulaError):
evaluate_formula(compile_formula("unknown_var"), {})
def test_division_by_zero_raises_formula_error(self):
with pytest.raises(FormulaError):
evaluate_formula(compile_formula("1 / 0"), {})
class TestFunctions:
def test_default_math_functions(self):
assert evaluate_formula(compile_formula("sqrt(16)"), {}) == 4.0
assert evaluate_formula(compile_formula("max(1, 2, 3)"), {}) == 3.0
assert evaluate_formula(compile_formula("min(1, 2, 3)"), {}) == 1.0
assert evaluate_formula(compile_formula("abs(-5)"), {}) == 5.0
assert evaluate_formula(compile_formula("exp(0)"), {}) == 1.0
assert math.isclose(evaluate_formula(compile_formula("log(exp(1))"), {}), 1.0)
def test_custom_injected_function(self):
formula = compile_formula('dep("power_density", "provides")')
result = evaluate_formula(
formula, {}, functions={"dep": lambda key, constraint_type: 99.0}
)
assert result == 99.0
def test_unknown_function_raises(self):
with pytest.raises(FormulaError):
evaluate_formula(compile_formula("unknown_fn(1)"), {})
def test_string_constant_passthrough_to_function(self):
formula = compile_formula('dep("mass")')
result = evaluate_formula(formula, {}, functions={"dep": lambda key: len(key)})
assert result == 4.0
class TestSecurity:
@pytest.mark.parametrize("source", [
"__import__('os').system('echo hi')",
"().__class__",
"[1, 2, 3]",
"{1: 2}",
"{1, 2}",
"(x for x in [1])",
"lambda: 1",
"1 if True else 0",
"1 == 1",
"x.__class__",
"x[0]",
"(lambda: 1)()",
"1; 2",
])
def test_disallowed_constructs_rejected(self, source):
with pytest.raises(FormulaError):
compile_formula(source)
def test_unregistered_function_name_never_executes(self):
"""exec/eval/__import__ etc. parse as ordinary Call nodes -- the
actual guarantee is that no function name is callable unless it's
explicitly in DEFAULT_FUNCTIONS or caller-supplied, checked at
evaluate time, not that the bare name is rejected at compile time."""
with pytest.raises(FormulaError):
evaluate_formula(compile_formula("exec('1')"), {})
def test_dunder_name_rejected(self):
with pytest.raises(FormulaError):
compile_formula("__builtins__")
def test_indirect_call_rejected(self):
with pytest.raises(FormulaError):
compile_formula("(a + b)(1)")
def test_invalid_syntax_raises_formula_error(self):
with pytest.raises(FormulaError):
compile_formula("2 +")
def test_boolean_constant_rejected(self):
with pytest.raises(FormulaError):
compile_formula("True")
def test_large_exponent_overflows_cleanly_not_hangs(self):
with pytest.raises(FormulaError):
evaluate_formula(compile_formula("9 ** 9 ** 9 ** 9"), {})

View File

@@ -0,0 +1,113 @@
"""End-to-end test that pass 2 can estimate a non-transport domain entirely
from domain-authored formulas, without touching the platform/actuator/
energy_storage physics model in Pipeline._estimate_physics."""
import pytest
from physcom.engine.constraint_resolver import ConstraintResolver
from physcom.engine.scorer import Scorer
from physcom.engine.pipeline import Pipeline
from physcom.models.domain import Domain, FreeVariable, MetricBound, MetricFormula
from physcom.models.entity import Dependency, Entity
def _build_archery_domain(repo):
repo.add_entity(Entity(
name="Recurve",
dimension="bow",
dependencies=[
Dependency("physical", "draw_weight", "20", None, "range_min"),
Dependency("physical", "draw_weight", "50", None, "range_max"),
],
))
repo.add_entity(Entity(
name="Carbon",
dimension="arrow",
dependencies=[
Dependency("physical", "arrow_mass", "0.02", None, "provides"),
],
))
return repo.add_domain(Domain(
name="archery_test",
metric_bounds=[
MetricBound("drawback_force", weight=0.6, norm_min=0, norm_max=100),
MetricBound("range", weight=0.4, norm_min=0, norm_max=300),
],
free_variables=[
FreeVariable(
name="draw_weight_chosen",
floor_formula='dep("draw_weight", "range_min")',
ceiling_formula='dep("draw_weight", "range_max")',
sort_order=0,
),
],
metric_formulas=[
MetricFormula(metric_name="drawback_force", formula="draw_weight_chosen * 2"),
MetricFormula(
metric_name="range",
formula='draw_weight_chosen * 5 / dep("arrow_mass")',
),
],
))
def test_formula_domain_scores_without_platform_actuator_shape(repo):
domain = _build_archery_domain(repo)
resolver = ConstraintResolver()
scorer = Scorer(domain)
pipeline = Pipeline(repo, resolver, scorer)
result = pipeline.run(
domain, ["bow", "arrow"], score_threshold=0.01, passes=[1, 2, 3, 5],
)
assert result.total_generated == 1
assert result.pass1_failed == 0
assert result.pass2_estimated == 1
assert result.pass3_above_threshold == 1
combos = repo.list_combinations()
assert len(combos) == 1
combo = combos[0]
scores = {
s["metric_name"]: s["raw_value"]
for s in repo.get_combination_scores(combo.id, domain.id)
}
# Both metrics increase monotonically with draw_weight_chosen and nothing
# trades off against it, so the optimizer should push to the declared
# ceiling (50) -- confirms _search_free_variables is actually searching,
# not just evaluating at the floor.
assert scores["drawback_force"] == pytest.approx(100.0, rel=0.02)
def test_formula_domain_zero_free_variables_direct_evaluation(repo):
"""A domain with metric_formulas but no free_variables should evaluate
each formula once directly -- no search loop at all."""
repo.add_entity(Entity(
name="Recurve",
dimension="bow",
dependencies=[Dependency("physical", "draw_weight", "30", None, "provides")],
))
repo.add_entity(Entity(name="Carbon", dimension="arrow"))
domain = repo.add_domain(Domain(
name="archery_direct_test",
metric_bounds=[MetricBound("drawback_force", weight=1.0, norm_min=0, norm_max=100)],
metric_formulas=[
MetricFormula(metric_name="drawback_force", formula='dep("draw_weight") * 2'),
],
))
resolver = ConstraintResolver()
scorer = Scorer(domain)
pipeline = Pipeline(repo, resolver, scorer)
result = pipeline.run(
domain, ["bow", "arrow"], score_threshold=0.01, passes=[1, 2, 3, 5],
)
assert result.pass2_estimated == 1
combos = repo.list_combinations()
scores = {
s["metric_name"]: s["raw_value"]
for s in repo.get_combination_scores(combos[0].id, domain.id)
}
assert scores["drawback_force"] == 60.0

View File

@@ -1,7 +1,7 @@
"""Tests for the database repository."""
from physcom.models.entity import Entity, Dependency
from physcom.models.domain import Domain, MetricBound
from physcom.models.domain import Domain, FreeVariable, MetricBound, MetricFormula
def test_ensure_dimension(repo):
@@ -63,6 +63,62 @@ def test_add_and_get_domain(repo):
assert loaded.metric_bounds[0].metric_name == "speed"
def test_add_domain_with_free_variables_and_formulas(repo):
domain = Domain(
name="archery_test",
metric_bounds=[MetricBound("drawback_force", weight=1.0, norm_min=0, norm_max=500)],
free_variables=[
FreeVariable(
name="draw_weight",
floor_formula='dep("draw_weight", "range_min")',
ceiling_formula='dep("draw_weight", "range_max")',
sort_order=0,
),
],
metric_formulas=[
MetricFormula(metric_name="drawback_force", formula="draw_weight * 1.5"),
],
)
saved = repo.add_domain(domain)
assert saved.id is not None
loaded = repo.get_domain("archery_test")
assert loaded is not None
assert len(loaded.free_variables) == 1
assert loaded.free_variables[0].name == "draw_weight"
assert loaded.free_variables[0].id is not None
assert len(loaded.metric_formulas) == 1
assert loaded.metric_formulas[0].formula == "draw_weight * 1.5"
def test_free_variable_and_formula_crud(repo):
domain = repo.add_domain(Domain(name="crud_test"))
fv = repo.add_free_variable(
domain.id,
FreeVariable(name="x", floor_formula="0", ceiling_formula="100", sort_order=0),
)
mf = repo.add_metric_formula(
domain.id, MetricFormula(metric_name="m", formula="x * 2")
)
repo.update_free_variable(
fv.id, FreeVariable(name="x", floor_formula="1", ceiling_formula="200", sort_order=0)
)
repo.update_metric_formula(mf.id, MetricFormula(metric_name="m", formula="x * 3"))
loaded = repo.get_domain_by_id(domain.id)
assert loaded.free_variables[0].floor_formula == "1"
assert loaded.free_variables[0].ceiling_formula == "200"
assert loaded.metric_formulas[0].formula == "x * 3"
repo.delete_free_variable(fv.id)
repo.delete_metric_formula(mf.id)
loaded = repo.get_domain_by_id(domain.id)
assert loaded.free_variables == []
assert loaded.metric_formulas == []
def test_combination_save_and_dedup(repo):
e1 = repo.add_entity(Entity(name="A", dimension="platform"))
e2 = repo.add_entity(Entity(name="B", dimension="actuator"))

View File

@@ -7,7 +7,7 @@ import pytest
from physcom.db.schema import init_db
from physcom.db.repository import Repository
from physcom.models.entity import Entity, Dependency
from physcom.models.domain import Domain, DomainConstraint, MetricBound
from physcom.models.domain import Domain, DomainConstraint, FreeVariable, MetricBound, MetricFormula
from physcom.models.combination import Combination
from physcom.snapshot import export_snapshot, import_snapshot
@@ -169,6 +169,46 @@ def test_import_with_combinations(seeded_repo, tmp_path):
assert len(fresh_combos) == len(data["combinations"])
def test_export_import_roundtrip_free_variables_and_formulas(repo, tmp_path):
domain = Domain(
name="archery_snapshot_test",
metric_bounds=[MetricBound("drawback_force", weight=1.0, norm_min=0, norm_max=100)],
free_variables=[
FreeVariable(
name="draw_weight_chosen",
floor_formula='dep("draw_weight", "range_min")',
ceiling_formula='dep("draw_weight", "range_max")',
sort_order=0,
),
],
metric_formulas=[
MetricFormula(metric_name="drawback_force", formula="draw_weight_chosen * 2"),
],
)
repo.add_domain(domain)
data = export_snapshot(repo)
exported = next(d for d in data["domains"] if d["name"] == "archery_snapshot_test")
assert exported["free_variables"] == [{
"name": "draw_weight_chosen", "sort_order": 0,
"floor_formula": 'dep("draw_weight", "range_min")',
"ceiling_formula": 'dep("draw_weight", "range_max")',
}]
assert exported["metric_formulas"] == [
{"metric_name": "drawback_force", "formula": "draw_weight_chosen * 2"},
]
conn = init_db(tmp_path / "fresh.db")
fresh = Repository(conn)
import_snapshot(fresh, data, clear=True)
loaded = fresh.get_domain("archery_snapshot_test")
assert len(loaded.free_variables) == 1
assert loaded.free_variables[0].floor_formula == 'dep("draw_weight", "range_min")'
assert len(loaded.metric_formulas) == 1
assert loaded.metric_formulas[0].formula == "draw_weight_chosen * 2"
def test_import_merge_skips_existing_domain(repo):
"""Merge import skips domains that already exist."""
domain = Domain(