From 3429bce8d0a725f88bb4509c0e2af62bd1b4b0dd Mon Sep 17 00:00:00 2001 From: Andrew Simonson Date: Sun, 16 Aug 2026 21:28:15 -0500 Subject: [PATCH] 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 --- src/physcom/db/repository.py | 86 ++++++++- src/physcom/db/schema.py | 18 ++ src/physcom/engine/constraint_resolver.py | 40 +++- src/physcom/engine/formula.py | 139 ++++++++++++++ src/physcom/engine/pipeline.py | 179 +++++++++++++++++- src/physcom/models/domain.py | 28 +++ src/physcom/snapshot.py | 33 +++- src/physcom_web/routes/domains.py | 95 +++++++++- .../templates/domains/_formulas_table.html | 56 ++++++ .../templates/domains/_free_vars_table.html | 64 +++++++ src/physcom_web/templates/domains/detail.html | 14 ++ tests/test_formula.py | 108 +++++++++++ tests/test_pipeline_formulas.py | 113 +++++++++++ tests/test_repository.py | 58 +++++- tests/test_snapshot.py | 42 +++- 15 files changed, 1064 insertions(+), 9 deletions(-) create mode 100644 src/physcom/engine/formula.py create mode 100644 src/physcom_web/templates/domains/_formulas_table.html create mode 100644 src/physcom_web/templates/domains/_free_vars_table.html create mode 100644 tests/test_formula.py create mode 100644 tests/test_pipeline_formulas.py diff --git a/src/physcom/db/repository.py b/src/physcom/db/repository.py index 1b1404b..b23786d 100644 --- a/src/physcom/db/repository.py +++ b/src/physcom/db/repository.py @@ -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") diff --git a/src/physcom/db/schema.py b/src/physcom/db/schema.py index 2e121d2..9cb70e0 100644 --- a/src/physcom/db/schema.py +++ b/src/physcom/db/schema.py @@ -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); diff --git a/src/physcom/engine/constraint_resolver.py b/src/physcom/engine/constraint_resolver.py index ed51a95..d5671ba 100644 --- a/src/physcom/engine/constraint_resolver.py +++ b/src/physcom/engine/constraint_resolver.py @@ -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: diff --git a/src/physcom/engine/formula.py b/src/physcom/engine/formula.py new file mode 100644 index 0000000..3bac6bb --- /dev/null +++ b/src/physcom/engine/formula.py @@ -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) diff --git a/src/physcom/engine/pipeline.py b/src/physcom/engine/pipeline.py index 5a39f82..e35be84 100644 --- a/src/physcom/engine/pipeline.py +++ b/src/physcom/engine/pipeline.py @@ -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, diff --git a/src/physcom/models/domain.py b/src/physcom/models/domain.py index b58b133..04d782e 100644 --- a/src/physcom/models/domain.py +++ b/src/physcom/models/domain.py @@ -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 diff --git a/src/physcom/snapshot.py b/src/physcom/snapshot.py index df0483f..c12bca2 100644 --- a/src/physcom/snapshot.py +++ b/src/physcom/snapshot.py @@ -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 diff --git a/src/physcom_web/routes/domains.py b/src/physcom_web/routes/domains.py index e7bb5e5..575e36b 100644 --- a/src/physcom_web/routes/domains.py +++ b/src/physcom_web/routes/domains.py @@ -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("//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("//free-vars//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("//free-vars//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("//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("//formulas//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("//formulas//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) diff --git a/src/physcom_web/templates/domains/_formulas_table.html b/src/physcom_web/templates/domains/_formulas_table.html new file mode 100644 index 0000000..58aab28 --- /dev/null +++ b/src/physcom_web/templates/domains/_formulas_table.html @@ -0,0 +1,56 @@ + + + + + + + + + + {% for mf in domain.metric_formulas %} + + + + + + + + + + + + + {% endfor %} + +
MetricFormula
{{ mf.metric_name }}{{ mf.formula }} + +
+ +
+
+ +

Add Formula

+
+
+ + + +
+
diff --git a/src/physcom_web/templates/domains/_free_vars_table.html b/src/physcom_web/templates/domains/_free_vars_table.html new file mode 100644 index 0000000..3476b79 --- /dev/null +++ b/src/physcom_web/templates/domains/_free_vars_table.html @@ -0,0 +1,64 @@ + + + + + + + + + + + + {% for fv in domain.free_variables %} + + + + + + + + + + + + + + + + + {% endfor %} + +
NameOrderFloor formulaCeiling formula
{{ fv.name }}{{ fv.sort_order }}{{ fv.floor_formula }}{{ fv.ceiling_formula }} + +
+ +
+
+ +

Add Free Variable

+
+
+ + + + + +
+
diff --git a/src/physcom_web/templates/domains/detail.html b/src/physcom_web/templates/domains/detail.html index 35810b4..9dda4e6 100644 --- a/src/physcom_web/templates/domains/detail.html +++ b/src/physcom_web/templates/domains/detail.html @@ -33,4 +33,18 @@
{% include "domains/_metrics_table.html" %}
+ +

Free Variables

+

Quantities pass 2's estimator searches to maximize this domain's composite score. Leave empty for domains where nothing needs sizing.

+ +
+ {% include "domains/_free_vars_table.html" %} +
+ +

Metric Formulas

+

How each metric's raw value is computed from declared entity properties (via dep(key, constraint_type="provides")) and any free variable above. If this domain declares any formulas, pass 2 uses them instead of the built-in vehicle physics model.

+ +
+ {% include "domains/_formulas_table.html" %} +
{% endblock %} diff --git a/tests/test_formula.py b/tests/test_formula.py new file mode 100644 index 0000000..0b2857e --- /dev/null +++ b/tests/test_formula.py @@ -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"), {}) diff --git a/tests/test_pipeline_formulas.py b/tests/test_pipeline_formulas.py new file mode 100644 index 0000000..02410c5 --- /dev/null +++ b/tests/test_pipeline_formulas.py @@ -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 diff --git a/tests/test_repository.py b/tests/test_repository.py index 418f1c0..e5c35b7 100644 --- a/tests/test_repository.py +++ b/tests/test_repository.py @@ -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")) diff --git a/tests/test_snapshot.py b/tests/test_snapshot.py index c8612ae..7e1b709 100644 --- a/tests/test_snapshot.py +++ b/tests/test_snapshot.py @@ -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(