Andrew Simonson
81b36e6bbe
Running a real phi4 pass-4 review pass surfaced a Rotorcraft + Gas
Turbine combo with a "speed" of 3,127 m/s (Mach 9), and the model's own
review text called it out directly: "unrealistic for urban commuting,
likely indicating an error." The reasoning for leaving air out of
DRAG_POWER_COEFF_BY_MEDIUM ("its L/D-based cruise model is already a
reasonable velocity-roughly-linear approximation") was wrong for the
same reason ground was wrong before: L/D-based drag force is also
roughly velocity-independent within a design cruise band, so it's still
just a mass-proportional constant with no v^2 term pushing back, and
inverting power/resistance for achieved speed had no ceiling there
either.
Added an aircraft-like reference cross-section (0.5 * rho_air * Cd(~0.2)
* frontal_area(~3.5 m^2)) to DRAG_POWER_COEFF_BY_MEDIUM for "air",
reusing the same closed-form cubic solve already built for ground.
Confirmed: the same combo's speed dropped from 3,127 m/s to 125.9 m/s
(a fast but physically plausible rotorcraft cruise), and a second phi4
pass-4 run on the corrected data no longer flags it -- the review now
discusses the speed score as a genuine strength instead of an apparent
error.
Water hull drag remains a known, unaddressed gap (would need its own
reference, not a car's or aircraft's frontal area).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1496 lines
70 KiB
Python
1496 lines
70 KiB
Python
"""Multi-pass pipeline orchestrator with incremental saves and resumability."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
|
|
from physcom.db.repository import Repository
|
|
from physcom.engine.combinator import generate_combinations
|
|
from physcom.engine.constraint_resolver import ConstraintResolver, ConstraintResult
|
|
from physcom.engine.scorer import Scorer, composite_score, normalize
|
|
from physcom.llm.base import LLMProvider, LLMRateLimitError
|
|
from physcom.llm.parsing import parse_rating
|
|
from physcom.models.combination import Combination, ScoredResult
|
|
from physcom.models.domain import Domain, MetricBound
|
|
from physcom.models.entity import Entity
|
|
|
|
# Stub-estimator heuristics (used only when no LLM provider is configured).
|
|
# Keyed by the same categorical vocabulary already used in seed data — never
|
|
# by entity name, so new entities inherit sensible behavior automatically.
|
|
|
|
THRUST_PROFILE_SAFETY: dict[str, float] = {
|
|
"low_continuous": 0.9,
|
|
"continuous_low": 0.9,
|
|
"moderate_continuous": 0.75,
|
|
"high_continuous": 0.6,
|
|
"extreme_continuous": 0.45,
|
|
"high_burst": 0.3,
|
|
"extreme_burst": 0.15,
|
|
}
|
|
|
|
# Baseline hazard of the energy form itself, independent of delivery profile.
|
|
ENERGY_FORM_SAFETY: dict[str, float] = {
|
|
"biological": 0.9,
|
|
"electrical": 0.85,
|
|
"wind": 0.9,
|
|
"gravitational": 0.9,
|
|
"radiation_pressure": 0.85,
|
|
"kinetic_stored": 0.7,
|
|
"pneumatic": 0.75,
|
|
"chemical_combustible": 0.6,
|
|
"ion_propellant": 0.75,
|
|
"chemical_propellant": 0.4,
|
|
"chemical_explosive": 0.3,
|
|
"nuclear_thermal": 0.35,
|
|
}
|
|
|
|
# How available the required infrastructure/fuel supply chain is today.
|
|
# Multiple matches in one combo (e.g. a platform's road_network requirement
|
|
# plus a storage's fuel_infrastructure requirement) are averaged.
|
|
INFRASTRUCTURE_AVAILABILITY: dict[tuple[str, str], float] = {
|
|
("road_network", "true"): 0.95,
|
|
("rail_network", "true"): 0.8,
|
|
("runway", "true"): 0.5,
|
|
("tow_or_winch", "true"): 0.5,
|
|
("hyperloop_tube", "true"): 0.1,
|
|
("launch_facility", "true"): 0.05,
|
|
("fuel_infrastructure", "none"): 1.0,
|
|
("fuel_infrastructure", "fuel_station"): 0.95,
|
|
("fuel_infrastructure", "charging_station"): 0.85,
|
|
("fuel_infrastructure", "cng_station"): 0.5,
|
|
("fuel_infrastructure", "coal_supply"): 0.5,
|
|
("fuel_infrastructure", "hydrogen_station"): 0.25,
|
|
("fuel_infrastructure", "compressed_air_station"): 0.3,
|
|
("fuel_infrastructure", "ammunition"): 0.3,
|
|
("fuel_infrastructure", "jet_fuel"): 0.6,
|
|
("fuel_infrastructure", "solid_propellant"): 0.15,
|
|
("fuel_infrastructure", "nuclear_fuel"): 0.05,
|
|
("fuel_infrastructure", "xenon_propellant"): 0.05,
|
|
}
|
|
|
|
# Crude freight-capacity proxy: kg of cargo per kg of vehicle structural
|
|
# mass. Was 500 -- a magnitude error (500x cargo-to-structure has no real
|
|
# vehicle analog). Real cargo ships run deadweight/lightship ratios of
|
|
# roughly 1.5-4x depending on class; 2.5 is a reasonable general-cargo
|
|
# midpoint for this domain-agnostic proxy.
|
|
CARGO_KG_PER_STRUCTURAL_KG: float = 2.5
|
|
|
|
# How mechanically proven/predictable an energy form is in practice — distinct
|
|
# from safety (risk when something goes wrong) and thrust_profile (delivery
|
|
# smoothness). Missing values fall back to a neutral 0.6.
|
|
ENERGY_FORM_RELIABILITY: dict[str, float] = {
|
|
"chemical_combustible": 0.85,
|
|
"electrical": 0.85,
|
|
"biological": 0.8,
|
|
"gravitational": 0.7,
|
|
"pneumatic": 0.7,
|
|
"kinetic_stored": 0.65,
|
|
"wind": 0.6,
|
|
"ion_propellant": 0.6,
|
|
"nuclear_thermal": 0.55,
|
|
"chemical_propellant": 0.5,
|
|
"radiation_pressure": 0.5,
|
|
"chemical_explosive": 0.45,
|
|
}
|
|
|
|
# ── power_density / range_fuel / cost_efficiency ──────────────────────
|
|
# These three use the platform's declared mass envelope as a combo-wide
|
|
# budget: every component (platform, actuator, storage) is bounded below
|
|
# by its own mass range_min, and the SUM is bounded above by the
|
|
# platform's mass range_max -- the exact aggregate check ConstraintResolver
|
|
# already performs in pass 1. The "leanest legal build" (every component
|
|
# at its own floor) is always a legal design point (pass 1 already
|
|
# validated it against the platform's ceiling), so it's used as the point
|
|
# estimate rather than an invented one.
|
|
|
|
# Radiation-pressure actuators (solar sails) don't declare a "mass" at
|
|
# all -- thrust scales with sail area, not carried mass -- so their
|
|
# effective mass is derived from declared footprint via a thin deployable
|
|
# sail film's areal density. Used the same way as BIOLOGICAL_OPERATOR_MASS_KG
|
|
# below: converts the entity's declared footprint FLOOR into a mass floor,
|
|
# not a fixed value -- above it, effective mass is a free, budget-competing
|
|
# variable like any other actuator (bigger sail = more collected power),
|
|
# sized by the same joint optimizer, not a one-off product spec.
|
|
SAIL_AREAL_DENSITY_KG_PER_M2: float = 0.05
|
|
|
|
# Human/animal actuators declare mass_min=0 (there's no minimum purchase
|
|
# quantity for a rider the way there is for an engine), but treated as a
|
|
# literal floor that lets the optimizer size a payload down toward 0kg of
|
|
# operator -- nonsensical, and it also breaks power = power_density * mass.
|
|
# Used as a FLOOR (not a fixed value) on top of the declared mass_min: at
|
|
# least one real operator must be present. Above that floor, actuator mass
|
|
# is a free, budget-competing, structurally-carried variable exactly like
|
|
# any mechanical actuator -- the "size" slider means more or bigger
|
|
# operators (a loaded cargo trike, a two-horse team), sized by the same
|
|
# joint optimizer everything else uses, not a fixed physiological constant.
|
|
BIOLOGICAL_OPERATOR_MASS_KG: dict[str, float] = {
|
|
"biological": 70.0, # one average human rider; Animal Traction shares this form too
|
|
}
|
|
|
|
# A platform's declared mass range often spans a whole real-world class, not
|
|
# one archetype -- Road Vehicle alone covers 50kg (motorcycle) to 36,000kg
|
|
# (truck). The floor is a legal build (pass 1 already checked it), but it's
|
|
# a motorcycle-scale build, not what a combo's own description usually
|
|
# implies. The geometric mean (not arithmetic) is the representative point
|
|
# for a range this wide: sqrt(50 * 36000) ~= 1343kg, in real commuter-car
|
|
# territory, versus the arithmetic mean (~18,000kg, a semi truck) or the
|
|
# floor (50kg, a motorcycle) -- real-world vehicle classes are far closer to
|
|
# log-uniformly distributed across a category than uniformly distributed.
|
|
def _representative_mass(mass_min: float, mass_max: float | None) -> float:
|
|
if mass_max and mass_min > 0:
|
|
return math.sqrt(mass_min * mass_max)
|
|
return max(mass_min, 100.0)
|
|
|
|
|
|
# Actuator + storage mass, sized to what's actually necessary rather than a
|
|
# fixed fraction of platform mass: enough actuator to sustain the
|
|
# platform's own performance requirement, enough storage to carry the
|
|
# domain's own "good" range target. Both share total_mass = p_rep + a + s,
|
|
# so the two requirements are coupled -- solved as a 2x2 linear system
|
|
# (Cramer's rule), not an iterative fit or an invented ratio:
|
|
#
|
|
# C1 * a = R1 * (p_rep + a + s) [a's capability meets requirement R1]
|
|
# C2 * s = R2 * (p_rep + a + s) [s's capability meets requirement R2]
|
|
#
|
|
# For the actuator equation, C1/R1 is either (specific_thrust, min_effective_accel)
|
|
# when the platform declares a real acceleration floor and the actuator
|
|
# declares real thrust (F=ma, both already exist in the seed data for
|
|
# aircraft/rocket combos -- no new data needed there), or (power_density,
|
|
# specific_energy_consumption * target_velocity) as the fallback -- "enough
|
|
# power to hold target_velocity against resistance" -- for every other
|
|
# platform, which needed one new attribute (target_velocity) since nothing
|
|
# in the schema previously declared a design speed for ground/water craft.
|
|
# For the storage equation, C2/R2 is always (energy_density, domain's own
|
|
# declared range_fuel norm_max * specific_energy_consumption) -- "enough
|
|
# energy to reach a genuinely good range for this domain," reusing the
|
|
# domain's own scoring ceiling rather than inventing a target.
|
|
#
|
|
# An infeasible system (the actuator is fundamentally too weak to ever
|
|
# reach the requirement, C1 <= R1) or a domain/platform missing the inputs
|
|
# it needs falls back to the entities' own bare floors -- a real
|
|
# limitation, not something to paper over with a default.
|
|
# Steady-state resistance (SPECIFIC_ENERGY_CONSUMPTION_J_PER_KG_M) only
|
|
# covers holding target_velocity -- real vehicles also carry reserve force
|
|
# for acceleration events (merging, passing, hills) that a pure cruise
|
|
# calculation would leave out entirely, which is why sizing off resistance
|
|
# alone undersizes the actuator relative to real vehicles. ~1.2 m/s^2 is a
|
|
# modest, real merging/passing acceleration capability, not a car's 0-60
|
|
# figure -- added directly to the resistance term below (see call site).
|
|
ACCELERATION_RESERVE_M_S2: float = 2.6
|
|
|
|
|
|
def _solve_two_requirement_masses(
|
|
p_rep: float, c1: float, r1: float, c2: float, r2: float,
|
|
a_min: float, s_min: float,
|
|
) -> tuple[float, float]:
|
|
a11, a12, b1 = c1 - r1, -r1, r1 * p_rep
|
|
a21, a22, b2 = -r2, c2 - r2, r2 * p_rep
|
|
det = a11 * a22 - a12 * a21
|
|
if abs(det) < 1e-9:
|
|
return a_min, s_min
|
|
a = (b1 * a22 - a12 * b2) / det
|
|
s = (a11 * b2 - a21 * b1) / det
|
|
if a <= 0 or s <= 0:
|
|
return a_min, s_min
|
|
return max(a, a_min), max(s, s_min)
|
|
|
|
|
|
def _solve_achievable_speed_mps(
|
|
power_density: float, floor_total: float, k_med: float, drag_coeff: float,
|
|
) -> float:
|
|
"""Invert specific_power = k_med*v + (drag_coeff/floor_total)*v^3 for v
|
|
-- the steady-state speed at which a build's actual power output exactly
|
|
balances mass-proportional resistance plus mass-independent aerodynamic
|
|
drag. A depressed cubic (no v^2 term) with drag_coeff/floor_total > 0
|
|
and k_med >= 0: A*v^3 + B*v - C = 0 is strictly increasing for v >= 0
|
|
(derivative 3*A*v^2 + B > 0 everywhere), so it has exactly one
|
|
non-negative real root -- solved directly via Cardano's formula, no
|
|
iteration needed. Falls back to the plain linear model (v = power/k_med)
|
|
when there's no drag coefficient for this medium, so ungraded media
|
|
behave exactly as before."""
|
|
if power_density <= 0 or k_med is None:
|
|
return 0.0
|
|
if drag_coeff <= 0 or floor_total <= 0:
|
|
return power_density / k_med if k_med else 0.0
|
|
A = drag_coeff / floor_total
|
|
B = k_med
|
|
C = power_density
|
|
p, q = B / A, -C / A
|
|
|
|
def cbrt(x: float) -> float:
|
|
return math.copysign(abs(x) ** (1 / 3), x) if x else 0.0
|
|
|
|
discriminant = (q / 2) ** 2 + (p / 3) ** 3 # always >= 0 given p, C >= 0
|
|
sqrt_disc = math.sqrt(discriminant)
|
|
v = cbrt(-q / 2 + sqrt_disc) + cbrt(-q / 2 - sqrt_disc)
|
|
return max(v, 0.0)
|
|
|
|
|
|
# Ambient energy forms (sun, wind, gravity) aren't a depletable onboard
|
|
# store the way a fuel tank is -- "distance before running out" doesn't
|
|
# apply (a sailboat doesn't run out of wind). Rather than degenerate to 0
|
|
# (mass_min=0, energy_density often undeclared entirely), range_fuel
|
|
# reports the domain's own declared ceiling for these: full marks is the
|
|
# physically honest answer, not an error. Food is deliberately NOT here:
|
|
# stopping to eat is a resupply, the same category as refuelling a tank,
|
|
# not a genuinely external/inexhaustible power source -- Biological Feed
|
|
# uses the normal storage-mass-limited range_fuel formula.
|
|
AMBIENT_ENERGY_FORMS: set[str] = {"wind", "radiation_pressure", "gravitational"}
|
|
|
|
# Resistive energy cost of travel, J per kg of vehicle per meter --
|
|
# rolling resistance for ground vehicles, cruise-flight lift/drag for
|
|
# aircraft, hull drag for water. Keyed by the platform's declared `medium`,
|
|
# not per-platform -- a real train's steel-wheel-on-rail is far more
|
|
# efficient than a car's tire, both currently "ground" -- flagged as the
|
|
# coarsest approximation here, same spot the earlier LLM comparison found
|
|
# every model's range_fuel guess off by 10-25x from real vehicles.
|
|
SPECIFIC_ENERGY_CONSUMPTION_J_PER_KG_M: dict[str, float] = {
|
|
"ground": 0.016 * 9.81, # combined rolling + aero "road load", Crr-equivalent ~ 0.016
|
|
"air": 9.81 / 10, # cruise flight, effective L/D ~ 10
|
|
"water": 0.05 * 9.81, # displacement-hull drag, rough order of magnitude
|
|
}
|
|
# Rocket-propelled (space medium) platforms aren't resistance-limited at
|
|
# all -- no drag to fight in vacuum -- so this "energy / (resistance *
|
|
# mass)" shape is the wrong model for them; real range is governed by the
|
|
# rocket equation (delta-v = exhaust velocity * ln(mass ratio)), which this
|
|
# pass does not implement. Space is deliberately left out of the dict above
|
|
# so it falls through to the old placeholder formula in the code below
|
|
# rather than silently claiming a resistance-based number that isn't real.
|
|
#
|
|
# FORMERLY A KNOWN GAP, now fixed below: the table above is mass-proportional
|
|
# resistance only (rolling resistance, effectively) -- no aerodynamic drag
|
|
# term (force ~ frontal_area * velocity^2, independent of mass). That's a
|
|
# reasonable approximation for something car-scale, where rolling resistance
|
|
# genuinely dominates at typical speeds and this was validated against real
|
|
# car range. It badly overestimated range for light/human-scale vehicles,
|
|
# where drag is the dominant resistance term and doesn't scale down with
|
|
# mass the way this formula assumes -- confirmed on a real combo (Light
|
|
# Personal Vehicle + Electric Motor + Rechargeable Battery, #876): a sane
|
|
# 9kg battery on a realistic 31kg vehicle came out to ~1,977km, a 6-9x
|
|
# overestimate against real e-bikes on comparable battery energy (~50-80km
|
|
# on ~500Wh). It also meant an achieved-speed metric derived from power
|
|
# alone (see DRAG_POWER_COEFF_BY_MEDIUM / _solve_achievable_speed_mps below)
|
|
# had no ceiling at all -- without a v^2-scaling force to push back, more
|
|
# power always bought proportionally more speed, forever.
|
|
#
|
|
# DRAG_POWER_COEFF_BY_MEDIUM below adds that missing term: a mass-INDEPENDENT
|
|
# drag power coefficient (0.5 * air_density * drag_coefficient * frontal_area,
|
|
# W per (m/s)^3) added on top of the existing mass-proportional term.
|
|
#
|
|
# "air" was originally left out of this table on the reasoning that its
|
|
# L/D-based cruise model was already a reasonable velocity-roughly-linear
|
|
# approximation -- true for computing energy per meter during a normal
|
|
# cruise, but WRONG for the exact same reason ground was wrong: L/D-based
|
|
# drag force is also roughly velocity-independent within a design cruise
|
|
# band, so it's still just "resistance = mass-proportional constant" with
|
|
# no v^2 term, and inverting power/resistance for achieved speed still had
|
|
# no ceiling. Confirmed live: a Rotorcraft + Gas Turbine combo showed a
|
|
# "speed" of 3,127 m/s (Mach 9) with phi4's pass-4 review flagging it
|
|
# directly ("unrealistic for urban commuting, likely indicating an
|
|
# error"). Added below with an aircraft-like reference cross-section.
|
|
# Water hull drag would need its own (different) treatment rather than
|
|
# reusing either reference, so it's left as a known remaining gap.
|
|
DRAG_POWER_COEFF_BY_MEDIUM: dict[str, float] = {
|
|
# 0.5 * rho_air(1.225 kg/m^3) * Cd(~0.3) * frontal_area(~2.2 m^2, small
|
|
# car reference) -- sanity check: at 30 m/s (108 km/h) this alone costs
|
|
# ~11kW, in the right ballpark for real highway cruise power.
|
|
"ground": 0.5 * 1.225 * 0.3 * 2.2,
|
|
# 0.5 * rho_air(1.225) * Cd(~0.2, streamlined fuselage) * frontal_area
|
|
# (~3.5 m^2, small aircraft/rotorcraft reference) -- sanity check: at
|
|
# 60 m/s (a fast urban rotorcraft cruise) this alone costs ~46kW, in
|
|
# the right ballpark for a light helicopter's real cruise power.
|
|
"air": 0.5 * 1.225 * 0.2 * 3.5,
|
|
}
|
|
|
|
# Structural manufacturing cost, $ per kg of platform mass -- certification
|
|
# and materials overhead scale hugely by medium (aerospace-grade vs.
|
|
# automotive steel vs. spacecraft-grade).
|
|
STRUCTURAL_COST_PER_KG_BY_MEDIUM: dict[str, float] = {
|
|
"ground": 8.0,
|
|
"air": 400.0,
|
|
"water": 15.0,
|
|
"space": 8000.0,
|
|
}
|
|
|
|
# Hardware manufacturing cost, $ per kg of actuator/storage-hardware mass,
|
|
# by energy form -- mature mass-produced tech (combustion, electric) is
|
|
# cheap per kg; exotic/regulated tech (nuclear, ion, rocket-grade) is not.
|
|
# biological is 0: there's no hardware to manufacture, the "actuator" is
|
|
# the operator's own body.
|
|
HARDWARE_COST_PER_KG_BY_ENERGY_FORM: dict[str, float] = {
|
|
"biological": 0.0,
|
|
"wind": 20.0,
|
|
"gravitational": 30.0,
|
|
"pneumatic": 35.0,
|
|
"chemical_combustible": 40.0,
|
|
"electrical": 60.0,
|
|
"kinetic_stored": 80.0,
|
|
"chemical_explosive": 150.0,
|
|
"chemical_propellant": 300.0,
|
|
"radiation_pressure": 500.0,
|
|
"ion_propellant": 5000.0,
|
|
"nuclear_thermal": 20000.0,
|
|
}
|
|
|
|
# Consumable energy price, $ per MJ delivered. Ambient sources (sun, wind,
|
|
# gravity) are genuinely free; food is a real recurring cost even though it
|
|
# isn't range-limiting -- cost and range are different questions, see
|
|
# AMBIENT_ENERGY_FORMS above. Replaces the old flat $/m ENERGY_FORM_BASE_COST
|
|
# placeholder with a real energy-priced figure.
|
|
FUEL_PRICE_PER_MJ: dict[str, float] = {
|
|
"wind": 0.0,
|
|
"gravitational": 0.0,
|
|
"radiation_pressure": 0.0,
|
|
"biological": 0.03,
|
|
"nuclear_thermal": 0.01,
|
|
"chemical_combustible": 0.04,
|
|
"electrical": 0.04,
|
|
"pneumatic": 0.02,
|
|
"kinetic_stored": 0.0,
|
|
"chemical_propellant": 1.0,
|
|
"chemical_explosive": 2.0,
|
|
"ion_propellant": 5.0,
|
|
}
|
|
|
|
# Total distance a vehicle travels over its operational life, used to
|
|
# amortize upfront/hardware cost into a $/m figure alongside operating
|
|
# cost. Coarse (per-medium, like the resistance table above) -- flagged as
|
|
# the same class of approximation.
|
|
LIFETIME_DISTANCE_M_BY_MEDIUM: dict[str, float] = {
|
|
"ground": 150_000_000.0,
|
|
"air": 3_000_000_000.0,
|
|
"water": 1_000_000_000.0,
|
|
"space": 5_000_000_000.0,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class _PhysicsContext:
|
|
"""Entity-level physics inputs for a combo that don't depend on a mass
|
|
allocation choice -- see Pipeline._physics_context."""
|
|
|
|
platform: Entity
|
|
actuator: Entity
|
|
storage: Entity
|
|
p_min: float
|
|
p_max: float | None
|
|
a_min: float
|
|
s_min: float
|
|
medium: str
|
|
actuator_energy_form: str | None
|
|
storage_energy_form: str | None
|
|
k_act: float
|
|
e_dens: float
|
|
k_med: float | None
|
|
p_rep: float
|
|
|
|
|
|
@dataclass
|
|
class PipelineResult:
|
|
"""Summary of a pipeline run."""
|
|
|
|
total_generated: int = 0
|
|
pass1_valid: int = 0
|
|
pass1_failed: int = 0
|
|
pass1_conditional: int = 0
|
|
pass2_estimated: int = 0
|
|
pass2_failed: int = 0
|
|
pass3_scored: int = 0
|
|
pass3_above_threshold: int = 0
|
|
pass3_failed: int = 0
|
|
pass4_reviewed: int = 0
|
|
pass4_failed: int = 0
|
|
pass5_human_reviewed: int = 0
|
|
top_results: list[dict] = field(default_factory=list)
|
|
|
|
|
|
class CancelledError(Exception):
|
|
"""Raised when a pipeline run is cancelled."""
|
|
|
|
|
|
def _describe_combination(combo: Combination) -> str:
|
|
"""Build a natural-language description of a combination."""
|
|
parts = [f"{e.dimension}: {e.name}" for e in combo.entities]
|
|
descriptions = [e.description for e in combo.entities if e.description]
|
|
header = " + ".join(parts)
|
|
detail = "; ".join(descriptions)
|
|
return f"{header}. {detail}"
|
|
|
|
|
|
class Pipeline:
|
|
"""Orchestrates the multi-pass viability pipeline."""
|
|
|
|
def __init__(
|
|
self,
|
|
repo: Repository,
|
|
resolver: ConstraintResolver,
|
|
scorer: Scorer,
|
|
llm: LLMProvider | None = None,
|
|
) -> None:
|
|
self.repo = repo
|
|
self.resolver = resolver
|
|
self.scorer = scorer
|
|
self.llm = llm
|
|
|
|
def _check_cancelled(self, run_id: int | None) -> None:
|
|
"""Raise CancelledError if the run has been cancelled."""
|
|
if run_id is None:
|
|
return
|
|
run = self.repo.get_pipeline_run(run_id)
|
|
if run and run["status"] == "cancelled":
|
|
raise CancelledError("Pipeline run cancelled")
|
|
|
|
def _update_run_counters(
|
|
self, run_id: int | None, result: PipelineResult, current_pass: int
|
|
) -> None:
|
|
"""Update pipeline_run progress counters in the DB."""
|
|
if run_id is None:
|
|
return
|
|
self.repo.update_pipeline_run(
|
|
run_id,
|
|
combos_pass1=result.pass1_valid
|
|
+ result.pass1_conditional
|
|
+ result.pass1_failed,
|
|
combos_pass2=result.pass2_estimated,
|
|
combos_pass3=result.pass3_scored,
|
|
combos_pass4=result.pass4_reviewed,
|
|
current_pass=current_pass,
|
|
)
|
|
|
|
def run(
|
|
self,
|
|
domain: Domain,
|
|
dimensions: list[str],
|
|
score_threshold: float = 0.1,
|
|
passes: list[int] | None = None,
|
|
run_id: int | None = None,
|
|
) -> PipelineResult:
|
|
if passes is None:
|
|
passes = [1, 2, 3, 4, 5]
|
|
|
|
result = PipelineResult()
|
|
|
|
# Mark run as running (unless already cancelled)
|
|
if run_id is not None:
|
|
run_record = self.repo.get_pipeline_run(run_id)
|
|
if run_record and run_record["status"] == "cancelled":
|
|
result.top_results = self.repo.get_top_results(domain.name, limit=20)
|
|
return result
|
|
self.repo.update_pipeline_run(
|
|
run_id,
|
|
status="running",
|
|
started_at=datetime.now(timezone.utc).isoformat(),
|
|
)
|
|
|
|
# Generate all combinations
|
|
combos = generate_combinations(self.repo, dimensions)
|
|
result.total_generated = len(combos)
|
|
|
|
# Save all combinations to DB (also loads status for existing combos).
|
|
# Deferred commit -- registering combos is instant/deterministic, so a
|
|
# crash here just means re-running the (cheap) registration loop, not
|
|
# losing anything worth protecting with a commit per row.
|
|
for combo in combos:
|
|
self.repo.save_combination(combo, commit=False)
|
|
self.repo.commit()
|
|
|
|
if run_id is not None:
|
|
self.repo.update_pipeline_run(run_id, total_combos=len(combos))
|
|
|
|
# Prepare metric lookup
|
|
bounds_by_name = {mb.metric_name: mb for mb in domain.metric_bounds}
|
|
|
|
# ── Phase-parallel: each pass runs to completion across every combo
|
|
# before the next pass starts, instead of walking each combo through
|
|
# every pass before moving to the next combo. This maximizes
|
|
# progress before the expensive/slow phase (pass 4's LLM calls) and
|
|
# keeps that phase's cost visible on its own, separate from the
|
|
# deterministic passes. It also removes any need for the two options
|
|
# to reconcile: pass 2 is estimator-only now (no LLM call in it at
|
|
# all -- self.llm is reserved for pass 4), so there's no combo that
|
|
# touches an LLM in both pass 2 and pass 4, and nothing here needs a
|
|
# live/resumed conversation across passes.
|
|
#
|
|
# Deterministic passes (1, 2, 3) defer commits and get flushed
|
|
# periodically + in `finally` below -- a crash there costs a cheap
|
|
# recompute, not lost work worth committing per write. Pass 4
|
|
# commits immediately after each call: those are slow and
|
|
# crash-prone (see the QwQ timeout saga), so that result is worth
|
|
# protecting the moment it lands.
|
|
combos_since_commit = 0
|
|
|
|
def _tick_commit() -> None:
|
|
nonlocal combos_since_commit
|
|
combos_since_commit += 1
|
|
if combos_since_commit >= 200:
|
|
self.repo.commit()
|
|
combos_since_commit = 0
|
|
|
|
try:
|
|
if 1 in passes:
|
|
for combo in combos:
|
|
self._check_cancelled(run_id)
|
|
_tick_commit()
|
|
self._process_pass1(combo, domain, result, run_id)
|
|
|
|
if 2 in passes:
|
|
for combo in combos:
|
|
self._check_cancelled(run_id)
|
|
_tick_commit()
|
|
self._process_pass2(combo, domain, bounds_by_name, result, run_id)
|
|
|
|
if 3 in passes:
|
|
for combo in combos:
|
|
self._check_cancelled(run_id)
|
|
_tick_commit()
|
|
self._process_pass3(
|
|
combo, domain, bounds_by_name, result, score_threshold, run_id
|
|
)
|
|
|
|
if 4 in passes and self.llm:
|
|
for combo in combos:
|
|
self._check_cancelled(run_id)
|
|
self._process_pass4(combo, domain, result, score_threshold, run_id)
|
|
|
|
except CancelledError:
|
|
if run_id is not None:
|
|
self.repo.update_pipeline_run(
|
|
run_id,
|
|
status="cancelled",
|
|
completed_at=datetime.now(timezone.utc).isoformat(),
|
|
)
|
|
result.top_results = self.repo.get_top_results(domain.name, limit=20)
|
|
return result
|
|
finally:
|
|
# Flush any batched deterministic writes -- runs on normal
|
|
# completion, cancellation, and any other exception propagating
|
|
# out of the loop, so nothing deferred above is ever silently lost
|
|
# on a clean exit path (a hard process crash is a different story
|
|
# and is exactly what the immediate LLM-call commits protect).
|
|
self.repo.commit()
|
|
|
|
# Mark run as completed
|
|
if run_id is not None:
|
|
self.repo.update_pipeline_run(
|
|
run_id,
|
|
status="completed",
|
|
completed_at=datetime.now(timezone.utc).isoformat(),
|
|
)
|
|
|
|
result.top_results = self.repo.get_top_results(domain.name, limit=20)
|
|
return result
|
|
|
|
@staticmethod
|
|
def _already_dead(combo: Combination, existing_result: dict | None) -> bool:
|
|
"""True if this combo is dead for every pass after 1 -- either a
|
|
generic failure (status ends in _fail) or a domain-specific block.
|
|
The domain-block case needs the extra existing_result check:
|
|
combo.status stays "valid" on purpose for it (domain-agnostic,
|
|
see _process_pass1's own comment on this), so pass_reached==1 with
|
|
the block already recorded is what actually marks it dead --
|
|
status alone isn't enough to catch it."""
|
|
if combo.status.endswith("_fail"):
|
|
return True
|
|
return bool(existing_result and existing_result["pass_reached"] == 1)
|
|
|
|
def _process_pass1(
|
|
self, combo: Combination, domain: Domain, result: PipelineResult, run_id: int | None
|
|
) -> None:
|
|
"""Constraint resolution for one combo. All writes deferred (commit=False)."""
|
|
existing_pass = self.repo.get_combo_pass_reached(combo.id, domain.id) or 0
|
|
if existing_pass >= 1:
|
|
if combo.status.endswith("_fail"):
|
|
result.pass1_failed += 1
|
|
return
|
|
existing_result = self.repo.get_existing_result(combo.id, domain.id)
|
|
if existing_result and existing_result["pass_reached"] == 1:
|
|
result.pass1_failed += 1
|
|
return
|
|
result.pass1_valid += 1
|
|
return
|
|
|
|
cr: ConstraintResult = self.resolver.resolve(combo)
|
|
if cr.status == "p1_fail":
|
|
combo.status = "p1_fail"
|
|
combo.block_reason = "; ".join(cr.violations)
|
|
self.repo.update_combination_status(
|
|
combo.id, "p1_fail", combo.block_reason, commit=False
|
|
)
|
|
# Save a result row so failed combos appear in results
|
|
self.repo.save_result(
|
|
combo.id, domain.id, composite_score=0.0, pass_reached=1, commit=False
|
|
)
|
|
result.pass1_failed += 1
|
|
self._update_run_counters(run_id, result, current_pass=1)
|
|
return
|
|
|
|
combo.status = "valid"
|
|
self.repo.update_combination_status(combo.id, "valid", commit=False)
|
|
|
|
# Domain constraint check (per-domain block only). combo.status stays
|
|
# "valid" here on purpose: it's domain-agnostic and the same combo can
|
|
# be blocked in this domain but valid in another. The per-domain
|
|
# block lives on combination_results.domain_block_reason (see
|
|
# count_combinations_by_status / get_all_results, which bucket on it).
|
|
if domain.constraints:
|
|
dc_result = self.resolver.check_domain_constraints(combo, domain.constraints)
|
|
if dc_result.status == "p1_fail":
|
|
self.repo.save_result(
|
|
combo.id, domain.id,
|
|
composite_score=0.0, pass_reached=1,
|
|
domain_block_reason="; ".join(dc_result.violations),
|
|
commit=False,
|
|
)
|
|
result.pass1_failed += 1
|
|
self._update_run_counters(run_id, result, current_pass=1)
|
|
return
|
|
|
|
if cr.status == "conditional":
|
|
result.pass1_conditional += 1
|
|
else:
|
|
result.pass1_valid += 1
|
|
self._update_run_counters(run_id, result, current_pass=1)
|
|
|
|
def _process_pass2(
|
|
self,
|
|
combo: Combination,
|
|
domain: Domain,
|
|
bounds_by_name: dict[str, MetricBound],
|
|
result: PipelineResult,
|
|
run_id: int | None,
|
|
) -> None:
|
|
"""Physics estimation for one combo. Estimator-only -- self.llm is
|
|
reserved for pass 4, never consulted here. All writes deferred."""
|
|
existing_result = self.repo.get_existing_result(combo.id, domain.id)
|
|
if self._already_dead(combo, existing_result):
|
|
return
|
|
existing_pass = self.repo.get_combo_pass_reached(combo.id, domain.id) or 0
|
|
if existing_pass >= 2:
|
|
result.pass2_estimated += 1
|
|
return
|
|
|
|
raw_metrics, feasible = self._stub_estimate(combo, domain.metric_bounds)
|
|
|
|
if not feasible:
|
|
# No platform mass within its own declared ceiling could
|
|
# structurally carry the required actuator+storage floor for
|
|
# this domain's performance targets -- power_density/range_fuel/
|
|
# cost_efficiency are per-kg ratios and don't naturally penalize
|
|
# that, so without this check a physically impossible build
|
|
# (an engine too big to fit on its own platform) could still
|
|
# score and pass. Domain-specific (the requirement floor depends
|
|
# on this domain's velocity/range targets), so this is a
|
|
# per-domain block like the domain-constraint check above, not
|
|
# a combo-wide p1_fail.
|
|
self.repo.save_result(
|
|
combo.id, domain.id, composite_score=0.0, pass_reached=1,
|
|
domain_block_reason=(
|
|
"Required actuator+storage mass exceeds what any platform "
|
|
"mass within its own declared ceiling could structurally "
|
|
"carry for this domain's performance targets"
|
|
),
|
|
commit=False,
|
|
)
|
|
result.pass1_failed += 1
|
|
self._update_run_counters(run_id, result, current_pass=2)
|
|
return
|
|
|
|
estimate_dicts = []
|
|
for mname, rval in raw_metrics.items():
|
|
mb = bounds_by_name.get(mname)
|
|
if mb and mb.metric_id:
|
|
estimate_dicts.append({
|
|
"metric_id": mb.metric_id,
|
|
"raw_value": rval,
|
|
"estimation_method": "stub",
|
|
"confidence": 1.0,
|
|
})
|
|
if estimate_dicts:
|
|
self.repo.save_raw_estimates(combo.id, domain.id, estimate_dicts, commit=False)
|
|
|
|
# Check for all-zero estimates → p2_fail
|
|
if raw_metrics and all(v == 0.0 for v in raw_metrics.values()):
|
|
combo.status = "p2_fail"
|
|
combo.block_reason = "All metric estimates are zero"
|
|
self.repo.update_combination_status(
|
|
combo.id, "p2_fail", combo.block_reason, commit=False
|
|
)
|
|
self.repo.save_result(
|
|
combo.id, domain.id, composite_score=0.0, pass_reached=2, commit=False
|
|
)
|
|
result.pass2_failed += 1
|
|
self._update_run_counters(run_id, result, current_pass=2)
|
|
return
|
|
|
|
result.pass2_estimated += 1
|
|
self._update_run_counters(run_id, result, current_pass=2)
|
|
|
|
def _process_pass3(
|
|
self,
|
|
combo: Combination,
|
|
domain: Domain,
|
|
bounds_by_name: dict[str, MetricBound],
|
|
result: PipelineResult,
|
|
score_threshold: float,
|
|
run_id: int | None,
|
|
) -> None:
|
|
"""Scoring for one combo. Reloads raw estimates from the DB (pass 2
|
|
ran as its own separate phase, not in-memory from this iteration).
|
|
All writes deferred."""
|
|
existing_result = self.repo.get_existing_result(combo.id, domain.id)
|
|
if self._already_dead(combo, existing_result):
|
|
return
|
|
existing_pass = self.repo.get_combo_pass_reached(combo.id, domain.id) or 0
|
|
if existing_pass >= 3:
|
|
result.pass3_scored += 1
|
|
if existing_result and existing_result["composite_score"] is not None:
|
|
if existing_result["composite_score"] >= score_threshold:
|
|
result.pass3_above_threshold += 1
|
|
return
|
|
|
|
existing_scores = self.repo.get_combination_scores(combo.id, domain.id)
|
|
raw_metrics = {s["metric_name"]: s["raw_value"] for s in existing_scores}
|
|
sr = self.scorer.score_combination(combo, raw_metrics)
|
|
|
|
score_dicts = []
|
|
for s in sr.scores:
|
|
mb = bounds_by_name.get(s.metric_name)
|
|
if mb and mb.metric_id:
|
|
score_dicts.append({
|
|
"metric_id": mb.metric_id,
|
|
"raw_value": s.raw_value,
|
|
"normalized_score": s.normalized_score,
|
|
"estimation_method": s.estimation_method,
|
|
"confidence": s.confidence,
|
|
})
|
|
if score_dicts:
|
|
self.repo.save_scores(combo.id, domain.id, score_dicts, commit=False)
|
|
|
|
# Preserve existing human data
|
|
novelty_flag = existing_result["novelty_flag"] if existing_result else None
|
|
human_notes = existing_result["human_notes"] if existing_result else None
|
|
|
|
if sr.composite_score < score_threshold:
|
|
self.repo.save_result(
|
|
combo.id, domain.id, sr.composite_score, pass_reached=3,
|
|
novelty_flag=novelty_flag, human_notes=human_notes, commit=False,
|
|
)
|
|
combo.status = "p3_fail"
|
|
combo.block_reason = (
|
|
f"Composite score {sr.composite_score:.4f} below threshold {score_threshold}"
|
|
)
|
|
self.repo.update_combination_status(
|
|
combo.id, "p3_fail", combo.block_reason, commit=False
|
|
)
|
|
result.pass3_failed += 1
|
|
result.pass3_scored += 1
|
|
self._update_run_counters(run_id, result, current_pass=3)
|
|
return
|
|
|
|
self.repo.save_result(
|
|
combo.id, domain.id, sr.composite_score, pass_reached=3,
|
|
novelty_flag=novelty_flag, human_notes=human_notes, commit=False,
|
|
)
|
|
self.repo.update_combination_status(combo.id, "scored", commit=False)
|
|
result.pass3_scored += 1
|
|
result.pass3_above_threshold += 1
|
|
self._update_run_counters(run_id, result, current_pass=3)
|
|
|
|
def _process_pass4(
|
|
self,
|
|
combo: Combination,
|
|
domain: Domain,
|
|
result: PipelineResult,
|
|
score_threshold: float,
|
|
run_id: int | None,
|
|
) -> None:
|
|
"""LLM plausibility review for one combo. Writes commit immediately
|
|
(default commit=True) -- slow, crash-prone calls worth protecting
|
|
the moment a result lands."""
|
|
cur_result = self.repo.get_existing_result(combo.id, domain.id)
|
|
if self._already_dead(combo, cur_result):
|
|
return
|
|
cur_pass = self.repo.get_combo_pass_reached(combo.id, domain.id) or 0
|
|
if cur_pass >= 4:
|
|
return
|
|
if not (
|
|
cur_result
|
|
and cur_result["composite_score"] is not None
|
|
and cur_result["composite_score"] >= score_threshold
|
|
):
|
|
return
|
|
|
|
description = _describe_combination(combo)
|
|
db_scores = self.repo.get_combination_scores(combo.id, domain.id)
|
|
score_dict = {
|
|
s["metric_name"]: s["normalized_score"]
|
|
for s in db_scores if s["normalized_score"] is not None
|
|
}
|
|
raw_dict = {
|
|
s["metric_name"]: s["raw_value"]
|
|
for s in db_scores if s["raw_value"] is not None
|
|
}
|
|
|
|
review_result: tuple[str, bool] | None = None
|
|
try:
|
|
review_result = self.llm.review_plausibility(
|
|
description, raw_dict, score_dict, domain
|
|
)
|
|
except LLMRateLimitError as exc:
|
|
self._wait_for_rate_limit(run_id, exc.retry_after)
|
|
try:
|
|
review_result = self.llm.review_plausibility(
|
|
description, raw_dict, score_dict, domain.metric_bounds
|
|
)
|
|
except LLMRateLimitError:
|
|
return # still limited; skip, retry next run
|
|
|
|
if review_result is None:
|
|
return
|
|
review_text, plausible = review_result
|
|
rating = parse_rating(review_text)
|
|
if not plausible:
|
|
self.repo.save_result(
|
|
combo.id, domain.id, cur_result["composite_score"], pass_reached=4,
|
|
novelty_flag=cur_result.get("novelty_flag"), llm_review=review_text,
|
|
human_notes=cur_result.get("human_notes"), qualitative_rating=rating,
|
|
)
|
|
combo.status = "p4_fail"
|
|
combo.block_reason = "LLM deemed implausible"
|
|
self.repo.update_combination_status(combo.id, "p4_fail", combo.block_reason)
|
|
result.pass4_failed += 1
|
|
else:
|
|
self.repo.save_result(
|
|
combo.id, domain.id, cur_result["composite_score"], pass_reached=4,
|
|
novelty_flag=cur_result.get("novelty_flag"), llm_review=review_text,
|
|
human_notes=cur_result.get("human_notes"), qualitative_rating=rating,
|
|
)
|
|
self.repo.update_combination_status(combo.id, "llm_reviewed")
|
|
result.pass4_reviewed += 1
|
|
self._update_run_counters(run_id, result, current_pass=4)
|
|
|
|
def _wait_for_rate_limit(self, run_id: int | None, retry_after: int) -> None:
|
|
"""Mark run rate_limited, sleep with cancel checks, then resume."""
|
|
if run_id is not None:
|
|
self.repo.update_pipeline_run(run_id, status="rate_limited")
|
|
waited = 0
|
|
while waited < retry_after:
|
|
time.sleep(5)
|
|
waited += 5
|
|
self._check_cancelled(run_id)
|
|
if run_id is not None:
|
|
self.repo.update_pipeline_run(run_id, status="running")
|
|
|
|
def _physics_context(
|
|
self, combo: Combination, bounds_by_name: dict[str, MetricBound]
|
|
) -> "_PhysicsContext | None":
|
|
"""Derive the entity-level physics inputs that don't depend on a
|
|
mass allocation choice -- shared by _stub_estimate (which picks the
|
|
allocation via solve or a special case) and _optimize_allocation
|
|
(which searches over candidate allocations). Returns None if the
|
|
combo doesn't have the platform/actuator/storage shape this whole
|
|
formula assumes (shouldn't happen for real combos, but a domain
|
|
without all three dimensions requested would hit this)."""
|
|
platform = next((e for e in combo.entities if e.dimension == "platform"), None)
|
|
actuator = next((e for e in combo.entities if e.dimension == "actuator"), None)
|
|
storage = next((e for e in combo.entities if e.dimension == "energy_storage"), None)
|
|
if platform is None or actuator is None or storage is None:
|
|
return None
|
|
|
|
def dep_value(entity, key, constraint_type) -> float | None:
|
|
for dep in entity.dependencies:
|
|
if dep.key == key and dep.constraint_type == constraint_type:
|
|
return float(dep.value)
|
|
return None
|
|
|
|
def dep_str(entity, key, constraint_type) -> str | None:
|
|
for dep in entity.dependencies:
|
|
if dep.key == key and dep.constraint_type == constraint_type:
|
|
return dep.value
|
|
return None
|
|
|
|
p_min = dep_value(platform, "mass", "range_min") or 0.0
|
|
a_min = dep_value(actuator, "mass", "range_min") or 0.0
|
|
s_min = dep_value(storage, "mass", "range_min") or 0.0
|
|
p_max = dep_value(platform, "mass", "range_max")
|
|
medium = dep_str(platform, "medium", "requires") or "ground"
|
|
|
|
return _PhysicsContext(
|
|
platform=platform, actuator=actuator, storage=storage,
|
|
p_min=p_min, p_max=p_max, a_min=a_min, s_min=s_min,
|
|
medium=medium,
|
|
actuator_energy_form=dep_str(actuator, "energy_form", "requires"),
|
|
storage_energy_form=dep_str(storage, "energy_form", "provides"),
|
|
k_act=dep_value(actuator, "power_density", "provides") or 0.0,
|
|
e_dens=dep_value(storage, "energy_density", "provides") or 0.0,
|
|
k_med=SPECIFIC_ENERGY_CONSUMPTION_J_PER_KG_M.get(medium),
|
|
p_rep=_representative_mass(p_min, p_max),
|
|
)
|
|
|
|
def _raw_physics_from_masses(
|
|
self,
|
|
ctx: "_PhysicsContext",
|
|
actuator_mass: float,
|
|
storage_mass: float,
|
|
bounds_by_name: dict[str, MetricBound],
|
|
units_by_name: dict[str, str],
|
|
platform_mass: float | None = None,
|
|
) -> dict[str, float]:
|
|
"""power_density/range_fuel/cost_efficiency/cargo_capacity for an
|
|
EXPLICIT mass allocation. `platform_mass` defaults to the
|
|
platform's representative mass (ctx.p_rep) -- pass an explicit
|
|
value to explore a specific weight class instead (see
|
|
evaluate_allocation). Cargo capacity is derived from THIS build's
|
|
actual platform+actuator mass (not a separate declared-floor
|
|
constant), with storage_mass subtracted out of that allowance --
|
|
see the deadweight/lightship comment at its computation below for
|
|
why fuel/battery competes with cargo instead of padding it. It
|
|
responds to the same optimizer/explore-slider choices every other
|
|
metric here does."""
|
|
p_mass = ctx.p_rep if platform_mass is None else platform_mass
|
|
out: dict[str, float] = {}
|
|
floor_total = p_mass + actuator_mass + storage_mass
|
|
power_density_value = (ctx.k_act * actuator_mass) / floor_total if floor_total else 0.0
|
|
|
|
if "power_density" in bounds_by_name:
|
|
out["power_density"] = power_density_value
|
|
|
|
# Deadweight/lightship cargo capacity. Real deadweight tonnage is a
|
|
# FIXED allowance sized off the vessel's own empty (lightship) mass
|
|
# -- hull + machinery, NOT fuel or cargo -- and fuel and cargo then
|
|
# SHARE that one allowance: a ship that bunkers more fuel has that
|
|
# much less room left for cargo, and vice versa. platform+actuator
|
|
# is the lightship analog here (the vehicle's own hardware);
|
|
# storage_mass is the fuel/battery competing with cargo for the
|
|
# same pool, not part of the base the pool is sized from -- get
|
|
# that backwards (basing the pool on platform+actuator+storage,
|
|
# as an earlier version of this did) and more battery looks like it
|
|
# BUYS more cargo room instead of using it up. Two ratio
|
|
# conventions coexist because heavy freight/maritime vehicles
|
|
# genuinely carry a much larger multiple of their own mass in
|
|
# cargo than light personal/delivery vehicles do (see
|
|
# CARGO_KG_PER_STRUCTURAL_KG's module comment); which one a domain
|
|
# scores is just which metric_name it declares. Floored at 0: a
|
|
# storage mass bigger than the whole allowance leaves no cargo
|
|
# room, not negative room.
|
|
lightship_mass = p_mass + actuator_mass
|
|
cargo_capacity_2_5x = max(0.0, lightship_mass * CARGO_KG_PER_STRUCTURAL_KG - storage_mass)
|
|
cargo_capacity_0_3x = max(0.0, lightship_mass * 0.3 - storage_mass)
|
|
if "cargo_capacity" in bounds_by_name:
|
|
out["cargo_capacity"] = cargo_capacity_2_5x
|
|
if "cargo_capacity_kg" in bounds_by_name:
|
|
out["cargo_capacity_kg"] = cargo_capacity_0_3x
|
|
|
|
# Achieved steady-state cruise speed, DERIVED from this specific
|
|
# build's actual power_density, the medium's mass-proportional
|
|
# resistance, and (ground/air, see DRAG_POWER_COEFF_BY_MEDIUM) a
|
|
# mass-independent aerodynamic drag term -- not a platform-declared
|
|
# constant. A build with more power than the platform's bare
|
|
# target_velocity requires achieves a genuinely higher speed here;
|
|
# an underbuilt one achieves less -- speed is an output of the
|
|
# build, not an input to it. Computed unconditionally (not just
|
|
# when "speed" is a scored metric) because range_fuel/cost_efficiency
|
|
# below both need it too: the energy actually spent per meter
|
|
# depends on how fast this build is actually going, drag included.
|
|
drag_coeff = DRAG_POWER_COEFF_BY_MEDIUM.get(ctx.medium, 0.0)
|
|
achieved_speed = _solve_achievable_speed_mps(power_density_value, floor_total, ctx.k_med, drag_coeff)
|
|
effective_k_med = (
|
|
(ctx.k_med + drag_coeff * achieved_speed ** 2 / floor_total)
|
|
if ctx.k_med is not None and floor_total > 0 else ctx.k_med
|
|
)
|
|
|
|
if "speed" in bounds_by_name:
|
|
out["speed"] = achieved_speed
|
|
|
|
if "range_fuel" in bounds_by_name:
|
|
if ctx.storage_energy_form in AMBIENT_ENERGY_FORMS or ctx.k_med is None:
|
|
mb = bounds_by_name.get("range_fuel")
|
|
out["range_fuel"] = mb.norm_max if mb else 0.0
|
|
elif floor_total > 0:
|
|
out["range_fuel"] = min(
|
|
(ctx.e_dens * storage_mass) / (effective_k_med * floor_total), 1e13
|
|
)
|
|
|
|
if "cost_efficiency" in bounds_by_name:
|
|
structural_cost = p_mass * STRUCTURAL_COST_PER_KG_BY_MEDIUM.get(
|
|
ctx.medium, STRUCTURAL_COST_PER_KG_BY_MEDIUM["ground"]
|
|
)
|
|
actuator_hw_cost = actuator_mass * HARDWARE_COST_PER_KG_BY_ENERGY_FORM.get(
|
|
ctx.actuator_energy_form, 50.0
|
|
)
|
|
storage_hw_cost = storage_mass * HARDWARE_COST_PER_KG_BY_ENERGY_FORM.get(
|
|
ctx.storage_energy_form, 50.0
|
|
)
|
|
upfront_cost = structural_cost + actuator_hw_cost + storage_hw_cost
|
|
lifetime_m = LIFETIME_DISTANCE_M_BY_MEDIUM.get(
|
|
ctx.medium, LIFETIME_DISTANCE_M_BY_MEDIUM["ground"]
|
|
)
|
|
amortized_per_m = upfront_cost / lifetime_m
|
|
|
|
fuel_price_per_mj = FUEL_PRICE_PER_MJ.get(ctx.storage_energy_form, 0.04)
|
|
energy_per_m_mj = (
|
|
(effective_k_med or SPECIFIC_ENERGY_CONSUMPTION_J_PER_KG_M["ground"]) * floor_total
|
|
) / 1e6
|
|
operating_per_m = energy_per_m_mj * fuel_price_per_mj
|
|
|
|
cost_per_m = amortized_per_m + operating_per_m
|
|
if units_by_name.get("cost_efficiency") == "$/(kg·m)":
|
|
# Divide by whichever cargo convention this domain actually
|
|
# scores, so cost-per-cargo-kg and the cargo_capacity number
|
|
# shown alongside it always agree; default to the heavy-
|
|
# vehicle ratio if a domain scores $/(kg·m) without scoring
|
|
# either cargo metric explicitly (matches prior behavior).
|
|
cargo_basis = out.get("cargo_capacity_kg", out.get("cargo_capacity", cargo_capacity_2_5x))
|
|
out["cost_efficiency"] = cost_per_m / max(cargo_basis, 1.0)
|
|
else:
|
|
out["cost_efficiency"] = cost_per_m
|
|
|
|
return out
|
|
|
|
def _decide_masses(
|
|
self,
|
|
ctx: "_PhysicsContext",
|
|
bounds_by_name: dict[str, MetricBound],
|
|
units_by_name: dict[str, str],
|
|
) -> tuple[float, float, float, bool]:
|
|
"""Pick the platform/actuator/storage mass for the build this domain
|
|
actually scores. First, the platform's declared physical
|
|
performance target (accel/thrust, or target_velocity/resistance)
|
|
sets a FLOOR -- a rotorcraft that can't produce enough thrust to
|
|
hover isn't a rotorcraft, regardless of how a smaller/cheaper
|
|
engine might score. Biological actuators (a rider's own body) and
|
|
radiation-pressure actuators (a solar sail) get the same treatment
|
|
with one addition: BIOLOGICAL_OPERATOR_MASS_KG / a footprint-derived
|
|
floor (see SAIL_AREAL_DENSITY_KG_PER_M2) sets a floor under the
|
|
floor -- at least one real operator, or the sail's own declared
|
|
minimum footprint, even if the performance-derived requirement
|
|
would otherwise ask for less -- but above that, mass is a free
|
|
variable exactly like a mechanical actuator's; "bigger" means more
|
|
or bigger operators, or a bigger sail, not a fixed constant. That
|
|
floor also
|
|
sets the smallest platform mass that could structurally carry it
|
|
(CARGO_KG_PER_STRUCTURAL_KG again, applied to the platform
|
|
carrying its own actuator+storage instead of cargo) -- below that,
|
|
no actuator/storage choice is physically possible. Above that
|
|
lower bound, platform mass is a real THIRD search variable, not
|
|
fixed at p_rep: a bigger platform also raises the structural cap
|
|
on how much actuator+storage it can carry, so growing all three
|
|
together can score higher than minimizing platform down to what's
|
|
merely required. Searched jointly (outer coarse-to-fine scan over
|
|
platform mass, inner coarse-to-fine scan over actuator/storage at
|
|
each candidate) for whatever allocation maximizes this domain's
|
|
own weighted composite score, using the same normalize()/
|
|
composite_score() the real scoring pass uses. Not "just enough to
|
|
function" and not "best score regardless of function" -- both,
|
|
floor then optimize jointly. Returns (actuator_mass, storage_mass,
|
|
platform_mass, feasible). `feasible` is False only when no platform
|
|
mass within its own declared ceiling could structurally carry the
|
|
required floor -- power_density/range_fuel/cost_efficiency are all
|
|
per-kg ratios, so they don't naturally penalize a build whose
|
|
absolute mass tramples its own platform's declared ceiling;
|
|
callers must treat an infeasible build as a hard fail rather than
|
|
trusting the (still-computable, still ratio-plausible) score. Also
|
|
used by evaluate_allocation to compute the slider's starting
|
|
values, so an explore session opens on the exact build the saved
|
|
score reflects.
|
|
"""
|
|
def dep_value(entity, key, constraint_type) -> float | None:
|
|
for dep in entity.dependencies:
|
|
if dep.key == key and dep.constraint_type == constraint_type:
|
|
return float(dep.value)
|
|
return None
|
|
|
|
# Step 1: the required floor (same solve as before -- now a floor
|
|
# for the search below, not the final answer).
|
|
min_accel = dep_value(ctx.platform, "min_effective_accel", "range_min")
|
|
specific_thrust = dep_value(ctx.actuator, "specific_thrust", "provides")
|
|
target_velocity = dep_value(ctx.platform, "target_velocity", "provides")
|
|
range_bounds = bounds_by_name.get("range_fuel")
|
|
target_range = range_bounds.norm_max if range_bounds else None
|
|
|
|
if min_accel and specific_thrust:
|
|
c1, r1 = specific_thrust, min_accel
|
|
elif target_velocity and ctx.k_med:
|
|
# Resistance alone (k_med) only covers steady-state cruise --
|
|
# a real vehicle also needs reserve force for acceleration
|
|
# events (merging, passing, hills), not just holding speed.
|
|
# F=ma: an acceleration reserve in m/s^2 is dimensionally a
|
|
# specific force (N/kg) exactly like k_med (J/(kg*m) = N/kg),
|
|
# so it adds directly before converting to specific power
|
|
# (P/mass = force/mass * v).
|
|
c1, r1 = ctx.k_act, (ctx.k_med + ACCELERATION_RESERVE_M_S2) * target_velocity
|
|
else:
|
|
c1 = r1 = 0.0 # no performance requirement available -- degenerates below
|
|
|
|
if target_range and ctx.k_med:
|
|
c2, r2 = ctx.e_dens, target_range * ctx.k_med
|
|
else:
|
|
c2 = r2 = 0.0
|
|
|
|
if c1 and c2:
|
|
required_actuator, required_storage = _solve_two_requirement_masses(
|
|
ctx.p_rep, c1, r1, c2, r2, ctx.a_min, ctx.s_min
|
|
)
|
|
else:
|
|
# No performance requirement available at all (e.g. a
|
|
# space-medium platform paired with an actuator that declares
|
|
# neither specific_thrust nor a usable target velocity) --
|
|
# fall back to bare floors, with the same near-zero-mass
|
|
# nominal reference used elsewhere so this doesn't silently
|
|
# degenerate to 0 power the way the original stub did.
|
|
required_actuator = ctx.a_min if ctx.a_min > 0.0 else 10.0
|
|
required_storage = ctx.s_min
|
|
|
|
if ctx.actuator_energy_form in BIOLOGICAL_OPERATOR_MASS_KG:
|
|
# At least one real operator, regardless of what the bare
|
|
# performance solve above would have asked for -- see the
|
|
# BIOLOGICAL_OPERATOR_MASS_KG module comment.
|
|
required_actuator = max(required_actuator, BIOLOGICAL_OPERATOR_MASS_KG[ctx.actuator_energy_form])
|
|
elif ctx.actuator_energy_form == "radiation_pressure":
|
|
# At least the entity's own declared minimum sail footprint,
|
|
# regardless of what the bare performance solve above would
|
|
# have asked for -- see the SAIL_AREAL_DENSITY_KG_PER_M2
|
|
# module comment.
|
|
footprint_floor = dep_value(ctx.actuator, "footprint", "range_min") or 0.0
|
|
required_actuator = max(required_actuator, footprint_floor * SAIL_AREAL_DENSITY_KG_PER_M2)
|
|
|
|
if ctx.p_max is None:
|
|
# No declared mass ceiling (e.g. Spaceship) -- no bounded
|
|
# budget to search within, use the requirement floor as-is.
|
|
return required_actuator, required_storage, ctx.p_rep, True
|
|
|
|
a_floor = max(ctx.a_min, required_actuator)
|
|
s_floor = max(ctx.s_min, required_storage)
|
|
structural_floor = a_floor + s_floor # min mass the platform must carry
|
|
|
|
# Platform mass is NOT just the required-floor minimum: because
|
|
# actuator+storage are capped at platform_mass * CARGO_KG_PER_STRUCTURAL_KG
|
|
# (see insufficient_structure), a bigger platform also buys room for
|
|
# a bigger, higher-scoring actuator/storage build -- so platform
|
|
# mass has to be searched jointly with them, not fixed. The lower
|
|
# bound still can't go below what's needed to carry the required
|
|
# floor at all (that's a physical requirement, not a scoring
|
|
# choice); p_rep is used only as the starting point for that
|
|
# search, not the answer.
|
|
p_lo = max(ctx.p_min, ctx.p_rep, structural_floor / CARGO_KG_PER_STRUCTURAL_KG)
|
|
p_lo = min(p_lo, ctx.p_max)
|
|
if p_lo * CARGO_KG_PER_STRUCTURAL_KG < structural_floor or ctx.p_max - p_lo < structural_floor:
|
|
# Even the smallest viable platform can't carry the required
|
|
# floor within the mass ceiling -- genuinely infeasible
|
|
# allocation, not a search problem. Best-effort fallback masses
|
|
# (feasible=False tells the caller not to trust the resulting
|
|
# score: power_density/range_fuel/cost_efficiency are all
|
|
# per-kg ratios, so they don't naturally penalize a build whose
|
|
# ABSOLUTE mass tramples its own platform's declared ceiling --
|
|
# something else has to catch that).
|
|
return a_floor, s_floor, p_lo, False
|
|
|
|
def objective(platform_mass: float, actuator_mass: float, storage_mass: float) -> float:
|
|
raw = self._raw_physics_from_masses(
|
|
ctx, actuator_mass, storage_mass,
|
|
bounds_by_name, units_by_name,
|
|
platform_mass=platform_mass,
|
|
)
|
|
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)
|
|
|
|
def best_at_platform(p: float, grid: int, rounds: int) -> tuple[float, float, float]:
|
|
budget = min(ctx.p_max - p, p * CARGO_KG_PER_STRUCTURAL_KG)
|
|
if budget < structural_floor:
|
|
return a_floor, s_floor, -1.0
|
|
return self._search_best_allocation(
|
|
a_floor, s_floor, budget, lambda a, s: objective(p, a, s), grid=grid, rounds=rounds,
|
|
)
|
|
|
|
# Outer coarse-to-fine search over platform mass. A cheap/low-res
|
|
# inner (a, s) search keeps every round affordable -- a low-res
|
|
# inner score is still a reasonable relative ranking of platform
|
|
# values even if each individual score isn't fully converged, and
|
|
# coarse-to-fine narrowing self-corrects across rounds. p_floor is
|
|
# the hard physical minimum and must never be narrowed past,
|
|
# unlike win_lo/win_hi which shrink each round.
|
|
p_floor = p_lo
|
|
win_lo, win_hi = p_lo, ctx.p_max
|
|
best_p, best_score = p_lo, -1.0
|
|
grid = 10
|
|
for _round in range(5):
|
|
for i in range(grid + 1):
|
|
p = win_lo + (win_hi - win_lo) * i / grid
|
|
if p < p_floor or p > ctx.p_max:
|
|
continue
|
|
_a, _s, sc = best_at_platform(p, grid=6, rounds=3)
|
|
if sc > best_score:
|
|
best_score, best_p = sc, p
|
|
span = max((win_hi - win_lo) / grid * 2, 1e-6)
|
|
win_lo = max(p_floor, best_p - span)
|
|
win_hi = min(ctx.p_max, best_p + span)
|
|
|
|
# Narrow local refinement at higher inner precision, over the same
|
|
# window the coarse scan above already settled into -- the coarse
|
|
# scan can land near, but not exactly on, the true optimum since
|
|
# it ranks platform values using a cheap inner search. A handful
|
|
# of medium-precision resamples of that same narrow window closes
|
|
# the gap without paying full precision at every one of the wide
|
|
# scan's many candidates.
|
|
for i in range(7):
|
|
p = win_lo + (win_hi - win_lo) * i / 6
|
|
if p < p_floor or p > ctx.p_max:
|
|
continue
|
|
_a, _s, sc = best_at_platform(p, grid=9, rounds=4)
|
|
if sc > best_score:
|
|
best_score, best_p = sc, p
|
|
|
|
actuator_mass, storage_mass, _score = best_at_platform(best_p, grid=12, rounds=6)
|
|
return actuator_mass, storage_mass, best_p, True
|
|
|
|
@staticmethod
|
|
def _search_best_allocation(
|
|
a_min: float, s_min: float, budget: float, objective,
|
|
grid: int = 12, rounds: int = 6,
|
|
) -> tuple[float, float, float]:
|
|
"""Coarse-to-fine grid search for the (actuator_mass, storage_mass)
|
|
that maximizes `objective` over the feasible triangle a>=a_min,
|
|
s>=s_min, a+s<=budget. No external dependency (scipy etc.) -- the
|
|
objective is smooth and low-dimensional enough that ~6 rounds of a
|
|
13x13 grid, narrowing the window each round, converges well in
|
|
well under a millisecond. `grid`/`rounds` are reduced by callers
|
|
doing many cheap scans (e.g. the outer platform-mass search in
|
|
_decide_masses) and left at their precise defaults for a final
|
|
answer."""
|
|
a_lo, a_hi = a_min, max(a_min, budget - s_min)
|
|
s_lo, s_hi = s_min, max(s_min, budget - a_min)
|
|
best_a, best_s, best_score = a_lo, s_lo, -1.0
|
|
|
|
for _round in range(rounds):
|
|
for i in range(grid + 1):
|
|
a = a_lo + (a_hi - a_lo) * i / grid
|
|
if a < a_min:
|
|
continue
|
|
s_cap = min(s_hi, budget - a)
|
|
if s_cap < s_min:
|
|
continue
|
|
for j in range(grid + 1):
|
|
s = s_lo + (s_cap - s_lo) * j / grid
|
|
if s < s_min:
|
|
continue
|
|
sc = objective(a, s)
|
|
if sc > best_score:
|
|
best_score, best_a, best_s = sc, a, s
|
|
a_span = max((a_hi - a_lo) / grid * 2, 1e-6)
|
|
s_span = max((s_hi - s_lo) / grid * 2, 1e-6)
|
|
a_lo, a_hi = max(a_min, best_a - a_span), min(budget - s_min, best_a + a_span)
|
|
s_lo, s_hi = max(s_min, best_s - s_span), min(budget - a_min, best_s + s_span)
|
|
|
|
return best_a, best_s, best_score
|
|
|
|
def _stub_estimate(
|
|
self, combo: Combination, metric_bounds: list[MetricBound]
|
|
) -> tuple[dict[str, float], bool]:
|
|
"""Deterministic estimation from declared entity attributes (no LLM).
|
|
|
|
power_density, range_fuel, and cost_efficiency are computed from the
|
|
platform's declared mass envelope treated as a combo-wide budget —
|
|
see the module-level comment above BIOLOGICAL_OPERATOR_MASS_KG for
|
|
the full formula rationale.
|
|
|
|
safety/availability/reliability/cargo_capacity/environmental_impact
|
|
are untouched — these are judgment calls (regulatory, economic,
|
|
qualitative), not physics, and stay on the categorical lookup-table
|
|
heuristics below (actuator's thrust_profile and energy_form and the
|
|
combo's infrastructure requirements).
|
|
|
|
cost_efficiency additionally checks the domain's declared unit:
|
|
"$/(kg·m)" (freight-style domains) isn't a rescaling of "$/m" — it's
|
|
a different quantity that needs dividing by cargo mass, not a
|
|
conversion factor.
|
|
|
|
Returns (raw_metrics, feasible). feasible is False when no platform
|
|
mass within its own declared ceiling could structurally carry the
|
|
required actuator+storage floor (see _decide_masses) -- raw_metrics
|
|
is still populated in that case (best-effort floor allocation) but
|
|
callers must not score it normally: none of power_density/
|
|
range_fuel/cost_efficiency are extensive quantities, so a build
|
|
whose absolute mass tramples its own platform's declared ceiling
|
|
can still produce perfectly plausible-looking per-kg ratios.
|
|
"""
|
|
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}
|
|
raw: dict[str, float] = {m: 0.0 for m in metric_names}
|
|
|
|
# Extract intrinsic properties from entities (unchanged — still
|
|
# drives the untouched blocks below).
|
|
power_density = 0.0 # W/kg
|
|
energy_density = 0.0 # J/kg
|
|
thrust_profile: str | None = None
|
|
energy_form: str | None = None
|
|
infra_matches: list[float] = []
|
|
for entity in combo.entities:
|
|
for dep in entity.dependencies:
|
|
if dep.key == "power_density" and dep.constraint_type == "provides":
|
|
power_density = max(power_density, float(dep.value))
|
|
if dep.key == "energy_density" and dep.constraint_type == "provides":
|
|
energy_density = max(energy_density, float(dep.value))
|
|
if dep.key == "thrust_profile" and dep.constraint_type == "provides":
|
|
thrust_profile = dep.value
|
|
if dep.key == "energy_form" and dep.constraint_type == "requires":
|
|
energy_form = dep.value
|
|
if dep.category == "infrastructure" and dep.constraint_type == "requires":
|
|
match = INFRASTRUCTURE_AVAILABILITY.get((dep.key, dep.value))
|
|
if match is not None:
|
|
infra_matches.append(match)
|
|
|
|
# ── platform/actuator/storage-specific extraction, for
|
|
# power_density / range_fuel / cost_efficiency / cargo_capacity
|
|
# only ─────────────────────────────────────────────────────────
|
|
ctx = self._physics_context(combo, bounds_by_name)
|
|
feasible = True
|
|
if ctx is not None:
|
|
actuator_mass, storage_mass, platform_mass, feasible = self._decide_masses(
|
|
ctx, bounds_by_name, units_by_name
|
|
)
|
|
raw.update(self._raw_physics_from_masses(
|
|
ctx, actuator_mass, storage_mass,
|
|
bounds_by_name, units_by_name,
|
|
platform_mass=platform_mass,
|
|
))
|
|
|
|
if "safety" in raw:
|
|
candidates = [
|
|
v for v in (
|
|
THRUST_PROFILE_SAFETY.get(thrust_profile),
|
|
ENERGY_FORM_SAFETY.get(energy_form),
|
|
)
|
|
if v is not None
|
|
]
|
|
raw["safety"] = min(candidates) if candidates else 0.6
|
|
|
|
if "availability" in raw:
|
|
raw["availability"] = (
|
|
sum(infra_matches) / len(infra_matches) if infra_matches else 0.5
|
|
)
|
|
|
|
if "range_degradation" in raw:
|
|
raw["range_degradation"] = 365 * 86400
|
|
|
|
# cargo_capacity / cargo_capacity_kg are set by _raw_physics_from_masses
|
|
# above, from the actual build mass -- not recomputed here.
|
|
|
|
if "environmental_impact" in raw:
|
|
raw["environmental_impact"] = max(0.0, power_density * 2e-7)
|
|
|
|
if "reliability" in raw:
|
|
raw["reliability"] = ENERGY_FORM_RELIABILITY.get(energy_form, 0.6)
|
|
|
|
return raw, feasible
|
|
|
|
def evaluate_allocation(
|
|
self,
|
|
combo: Combination,
|
|
domain: Domain,
|
|
platform_mass: float | None = None,
|
|
actuator_mass: float | None = None,
|
|
storage_mass: float | None = None,
|
|
) -> dict | None:
|
|
"""Direct, non-optimizing exploration: compute the resulting raw
|
|
metrics, normalized scores, and composite score for an EXPLICIT
|
|
(platform, actuator, storage) mass choice -- "what happens to
|
|
range and power if I build a bigger motor, or pick a heavier
|
|
weight class," not "what's the best possible build." Exists
|
|
purely for exploration (a combo detail page slider); nothing here
|
|
is ever persisted.
|
|
|
|
Any mass left as None defaults to what the real requirement-based
|
|
solve already picked (see _decide_masses / _stub_estimate), so a
|
|
slider opens on today's actual build, not an arbitrary point.
|
|
Explicit values are floor-clamped to each component's own declared
|
|
minimum (platform is also ceiling-clamped to its declared max) --
|
|
never silently allowed below what pass 1 would have rejected.
|
|
|
|
Returns None only for combos with no declared platform mass
|
|
ceiling to bound a weight-class slider (e.g. Spaceship). Every
|
|
actuator type gets sliders, including biological (rider/operator
|
|
mass, see BIOLOGICAL_OPERATOR_MASS_KG) and radiation-pressure
|
|
(effective sail mass derived from footprint, see
|
|
SAIL_AREAL_DENSITY_KG_PER_M2) -- both are real, budget-competing
|
|
variables like any mechanical actuator's mass.
|
|
"""
|
|
bounds_by_name = {mb.metric_name: mb for mb in domain.metric_bounds}
|
|
units_by_name = {mb.metric_name: mb.unit for mb in domain.metric_bounds}
|
|
ctx = self._physics_context(combo, bounds_by_name)
|
|
if ctx is None or ctx.p_max is None:
|
|
return None
|
|
|
|
default_actuator, default_storage, default_platform, _feasible = self._decide_masses(
|
|
ctx, bounds_by_name, units_by_name
|
|
)
|
|
p_mass = default_platform if platform_mass is None else platform_mass
|
|
a_mass = default_actuator if actuator_mass is None else actuator_mass
|
|
s_mass = default_storage if storage_mass is None else storage_mass
|
|
|
|
p_mass = max(ctx.p_min, min(p_mass, ctx.p_max))
|
|
a_mass = max(ctx.a_min, a_mass)
|
|
s_mass = max(ctx.s_min, s_mass)
|
|
|
|
raw = self._raw_physics_from_masses(
|
|
ctx, a_mass, s_mass,
|
|
bounds_by_name, units_by_name,
|
|
platform_mass=p_mass,
|
|
)
|
|
|
|
normalized: dict[str, float] = {}
|
|
scores, weights = [], []
|
|
for mb in domain.metric_bounds:
|
|
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
|
|
normalized[mb.metric_name] = n
|
|
scores.append(n)
|
|
weights.append(mb.weight)
|
|
|
|
# Loose slider ceilings for the UI: how big this component could
|
|
# get if platform and the other component sat at their own floors
|
|
# -- not a hard physics limit, just a sane default range to draw.
|
|
actuator_slider_max = max(a_mass, ctx.p_max - ctx.p_min - ctx.s_min)
|
|
storage_slider_max = max(s_mass, ctx.p_max - ctx.p_min - ctx.a_min)
|
|
|
|
total_mass = p_mass + a_mass + s_mass
|
|
return {
|
|
"platform_mass": p_mass, "platform_min": ctx.p_min, "platform_max": ctx.p_max,
|
|
"actuator_mass": a_mass, "actuator_min": ctx.a_min, "actuator_slider_max": actuator_slider_max,
|
|
"storage_mass": s_mass, "storage_min": ctx.s_min, "storage_slider_max": storage_slider_max,
|
|
"total_mass": total_mass,
|
|
# The sliders are intentionally loose (see actuator/storage_slider_max
|
|
# above) so exploration isn't boxed in by wherever the platform slider
|
|
# currently sits. That means a chosen build can exceed the platform's
|
|
# own declared mass ceiling -- physically, more assembled mass than
|
|
# this platform category is rated to carry. Flagged, not blocked.
|
|
"exceeds_platform_envelope": total_mass > ctx.p_max,
|
|
# None of power_density/range_fuel/cost_efficiency penalize a
|
|
# platform mass that's too small to structurally carry its own
|
|
# actuator+storage -- they only see the total. Reuse the same
|
|
# structure-supports-N-times-its-own-mass ratio already used for
|
|
# cargo_capacity_kg (CARGO_KG_PER_STRUCTURAL_KG) rather than a
|
|
# one-off constant: a platform can't carry more actuator+storage
|
|
# mass than that, any more than it could carry that much cargo.
|
|
"insufficient_structure": (a_mass + s_mass) > p_mass * CARGO_KG_PER_STRUCTURAL_KG,
|
|
"raw_metrics": raw,
|
|
"normalized_scores": normalized,
|
|
"composite_score": composite_score(scores, weights),
|
|
}
|