"""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) # 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. # # KNOWN GAP: this whole table is mass-proportional resistance only (rolling # resistance, effectively) -- there's 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 overestimates 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). The mass # allocation itself was fine (correctly floor-clamped, nothing oversized) -- # this is a missing term in the resistance formula, not an allocation bug, # so a mass-allocation optimizer wouldn't fix it either. Real fix needs a # genuine drag term (frontal-area-ish figure -- `footprint` exists but is a # ground-footprint number, not obviously the right proxy for cross-sectional # area facing the wind -- and a drag coefficient assumption), scoped # separately from the resistance-constant tuning already done here. # 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], cargo_capacity_kg: float, platform_mass: float | None = None, ) -> dict[str, float]: """power_density/range_fuel/cost_efficiency 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).""" 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 if "power_density" in bounds_by_name: out["power_density"] = (ctx.k_act * actuator_mass) / floor_total if floor_total else 0.0 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) / (ctx.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 = ( (ctx.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)": out["cost_efficiency"] = cost_per_m / max(cargo_capacity_kg, 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], cargo_capacity_kg: float, ) -> 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, cargo_capacity_kg, 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 mass_total = 0.0 # kg, extensive — components share one vehicle 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 == "mass" and dep.constraint_type == "range_min": mass_total += 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) mass = mass_total if mass_total > 0 else 100.0 # kg, default if undeclared cargo_capacity_kg = mass * CARGO_KG_PER_STRUCTURAL_KG # ── platform/actuator/storage-specific extraction, for # power_density / range_fuel / cost_efficiency 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, cargo_capacity_kg ) raw.update(self._raw_physics_from_masses( ctx, actuator_mass, storage_mass, bounds_by_name, units_by_name, cargo_capacity_kg, 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 if "cargo_capacity" in raw: raw["cargo_capacity"] = cargo_capacity_kg if "cargo_capacity_kg" in raw: raw["cargo_capacity_kg"] = mass * 0.3 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 cargo_capacity_kg = (ctx.p_min + ctx.a_min + ctx.s_min) * CARGO_KG_PER_STRUCTURAL_KG default_actuator, default_storage, default_platform, _feasible = self._decide_masses( ctx, bounds_by_name, units_by_name, cargo_capacity_kg ) 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, cargo_capacity_kg, 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), }